diff --git a/libs/dingtalk_api/api.py b/libs/dingtalk_api/api.py
index d1c7065f..3d483a3a 100644
--- a/libs/dingtalk_api/api.py
+++ b/libs/dingtalk_api/api.py
@@ -3,7 +3,6 @@ import json
import time
from typing import Callable
import dingtalk_stream # type: ignore
-from dingtalk_stream import AckMessage, ChatbotHandler, CallbackHandler, CallbackMessage, ChatbotMessage, AICardReplier
from .EchoHandler import EchoTextHandler
from .dingtalkevent import DingTalkEvent
import httpx
@@ -254,24 +253,23 @@ class DingTalkClient:
await self.logger.error(f'failed to send proactive massage to group: {traceback.format_exc()}')
raise Exception(f'failed to send proactive massage to group: {traceback.format_exc()}')
- async def create_and_card(self, temp_card_id: str, incoming_message: dingtalk_stream.ChatbotMessage,quote_origin:bool=False):
- content_key = "content"
- card_data = {content_key: ""}
+ async def create_and_card(
+ self, temp_card_id: str, incoming_message: dingtalk_stream.ChatbotMessage, quote_origin: bool = False
+ ):
+ content_key = 'content'
+ card_data = {content_key: ''}
- card_instance = dingtalk_stream.AICardReplier(
- self.client, incoming_message
- )
+ card_instance = dingtalk_stream.AICardReplier(self.client, incoming_message)
# print(card_instance)
# 先投放卡片: https://open.dingtalk.com/document/orgapp/create-and-deliver-cards
card_instance_id = await card_instance.async_create_and_deliver_card(
- temp_card_id, card_data,
+ temp_card_id,
+ card_data,
)
- return card_instance,card_instance_id
+ return card_instance, card_instance_id
- async def send_card_message(self,
- card_instance,
- card_instance_id: str,content: str,is_final: bool):
- content_key = "content"
+ async def send_card_message(self, card_instance, card_instance_id: str, content: str, is_final: bool):
+ content_key = 'content'
try:
await card_instance.async_streaming(
card_instance_id,
@@ -286,16 +284,12 @@ class DingTalkClient:
await card_instance.async_streaming(
card_instance_id,
content_key=content_key,
- content_value="",
+ content_value='',
append=False,
finished=is_final,
failed=True,
)
-
-
-
-
async def start(self):
"""启动 WebSocket 连接，监听消息"""
await self.client.start()
diff --git a/pkg/api/http/controller/groups/pipelines/webchat.py b/pkg/api/http/controller/groups/pipelines/webchat.py
index 6dc7f85a..c094731b 100644
--- a/pkg/api/http/controller/groups/pipelines/webchat.py
+++ b/pkg/api/http/controller/groups/pipelines/webchat.py
@@ -14,8 +14,9 @@ class WebChatDebugRouterGroup(group.RouterGroup):
async def stream_generator(generator):
async for message in generator:
- yield f"data: {json.dumps({'message': message})}\n\n"
- yield "data: {\"type\": \"end\"}\n\n"
+ yield f'data: {json.dumps({"message": message})}\n\n'
+ yield 'data: {"type": "end"}\n\n'
+
try:
data = await quart.request.get_json()
session_type = data.get('session_type', 'person')
@@ -34,18 +35,18 @@ class WebChatDebugRouterGroup(group.RouterGroup):
return self.http_status(404, -1, 'WebChat adapter not found')
if is_stream:
-
- generator = webchat_adapter.send_webchat_message(pipeline_uuid, session_type, message_chain_obj, is_stream)
-
- return quart.Response(
- stream_generator(generator),
- mimetype='text/event-stream'
+ generator = webchat_adapter.send_webchat_message(
+ pipeline_uuid, session_type, message_chain_obj, is_stream
)
+ return quart.Response(stream_generator(generator), mimetype='text/event-stream')
+
else:
# result = await webchat_adapter.send_webchat_message(pipeline_uuid, session_type, message_chain_obj)
result = None
- async for message in webchat_adapter.send_webchat_message(pipeline_uuid, session_type, message_chain_obj):
+ async for message in webchat_adapter.send_webchat_message(
+ pipeline_uuid, session_type, message_chain_obj
+ ):
result = message
if result is not None:
return self.success(
@@ -56,7 +57,6 @@ class WebChatDebugRouterGroup(group.RouterGroup):
else:
return self.http_status(400, -1, 'message is required')
-
except Exception as e:
return self.http_status(500, -1, f'Internal server error: {str(e)}')
diff --git a/pkg/core/entities.py b/pkg/core/entities.py
index 31514fa8..5f357d78 100644
--- a/pkg/core/entities.py
+++ b/pkg/core/entities.py
@@ -87,7 +87,9 @@ class Query(pydantic.BaseModel):
"""使用的函数，由前置处理器阶段设置"""
resp_messages: (
- typing.Optional[list[llm_entities.Message]] | typing.Optional[list[platform_message.MessageChain]] | typing.Optional[list[llm_entities.MessageChunk]]
+ typing.Optional[list[llm_entities.Message]]
+ | typing.Optional[list[platform_message.MessageChain]]
+ | typing.Optional[list[llm_entities.MessageChunk]]
) = []
"""由Process阶段生成的回复消息对象列表"""
diff --git a/pkg/pipeline/cntfilter/cntfilter.py b/pkg/pipeline/cntfilter/cntfilter.py
index 0bbc5103..e035c1d0 100644
--- a/pkg/pipeline/cntfilter/cntfilter.py
+++ b/pkg/pipeline/cntfilter/cntfilter.py
@@ -67,7 +67,7 @@ class ContentFilterStage(stage.PipelineStage):
if query.pipeline_config['safety']['content-filter']['scope'] == 'output-msg':
return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
if not message.strip():
- return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+ return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
else:
for filter in self.filter_chain:
if filter_entities.EnableStage.PRE in filter.enable_stages:
diff --git a/pkg/pipeline/process/handlers/chat.py b/pkg/pipeline/process/handlers/chat.py
index 0f802658..6c428473 100644
--- a/pkg/pipeline/process/handlers/chat.py
+++ b/pkg/pipeline/process/handlers/chat.py
@@ -81,9 +81,7 @@ class ChatMessageHandler(handler.MessageHandler):
query.resp_message_chain.pop()
query.resp_messages.append(result)
- self.ap.logger.info(
- f'对话({query.query_id})流式响应: {self.cut_str(result.readable_str())}'
- )
+ self.ap.logger.info(f'对话({query.query_id})流式响应: {self.cut_str(result.readable_str())}')
if result.content is not None:
text_length += len(result.content)
diff --git a/pkg/pipeline/respback/respback.py b/pkg/pipeline/respback/respback.py
index f4153218..c7824856 100644
--- a/pkg/pipeline/respback/respback.py
+++ b/pkg/pipeline/respback/respback.py
@@ -3,12 +3,10 @@ from __future__ import annotations
import random
import asyncio
-from typing_inspection.typing_objects import is_final
from ...platform.types import events as platform_events
from ...platform.types import message as platform_message
-from ...provider import entities as llm_entities
from .. import stage, entities
from ...core import entities as core_entities
@@ -56,6 +54,4 @@ class SendResponseBackStage(stage.PipelineStage):
quote_origin=quote_origin,
)
-
-
return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
diff --git a/pkg/platform/adapter.py b/pkg/platform/adapter.py
index e4369efb..3412be3c 100644
--- a/pkg/platform/adapter.py
+++ b/pkg/platform/adapter.py
@@ -25,7 +25,6 @@ class MessagePlatformAdapter(metaclass=abc.ABCMeta):
logger: EventLogger
-
def __init__(self, config: dict, ap: app.Application, logger: EventLogger):
"""初始化适配器
@@ -80,12 +79,12 @@ class MessagePlatformAdapter(metaclass=abc.ABCMeta):
"""
raise NotImplementedError
- async def create_message_card(self, message_id:typing.Type[str,int], event:platform_events.MessageEvent) -> bool:
+ async def create_message_card(self, message_id: typing.Type[str, int], event: platform_events.MessageEvent) -> bool:
"""创建卡片消息
Args:
message_id (str): 消息ID
event (platform_events.MessageEvent): 消息源事件
- """
+ """
return False
async def is_muted(self, group_id: int) -> bool:
@@ -94,8 +93,8 @@ class MessagePlatformAdapter(metaclass=abc.ABCMeta):
def register_listener(
self,
- event_type: typing.Type[platform_message.Event],
- callback: typing.Callable[[platform_message.Event, MessagePlatformAdapter], None],
+ event_type: typing.Type[platform_events.Event],
+ callback: typing.Callable[[platform_events.Event, MessagePlatformAdapter], None],
):
"""注册事件监听器
@@ -107,8 +106,8 @@ class MessagePlatformAdapter(metaclass=abc.ABCMeta):
def unregister_listener(
self,
- event_type: typing.Type[platform_message.Event],
- callback: typing.Callable[[platform_message.Event, MessagePlatformAdapter], None],
+ event_type: typing.Type[platform_events.Event],
+ callback: typing.Callable[[platform_events.Event, MessagePlatformAdapter], None],
):
"""注销事件监听器
@@ -167,7 +166,7 @@ class EventConverter:
"""事件转换器基类"""
@staticmethod
- def yiri2target(event: typing.Type[platform_message.Event]):
+ def yiri2target(event: typing.Type[platform_events.Event]):
"""将源平台事件转换为目标平台事件
Args:
@@ -179,7 +178,7 @@ class EventConverter:
raise NotImplementedError
@staticmethod
- def target2yiri(event: typing.Any) -> platform_message.Event:
+ def target2yiri(event: typing.Any) -> platform_events.Event:
"""将目标平台事件的调用参数转换为源平台的事件参数对象
Args:
diff --git a/pkg/platform/sources/dingtalk.py b/pkg/platform/sources/dingtalk.py
index 9f834f2a..8bd6e187 100644
--- a/pkg/platform/sources/dingtalk.py
+++ b/pkg/platform/sources/dingtalk.py
@@ -149,10 +149,10 @@ class DingTalkAdapter(adapter.MessagePlatformAdapter):
quote_origin: bool = False,
is_final: bool = False,
):
- event = await DingTalkEventConverter.yiri2target(
- message_source,
- )
- incoming_message = event.incoming_message
+ # event = await DingTalkEventConverter.yiri2target(
+ # message_source,
+ # )
+ # incoming_message = event.incoming_message
# msg_id = incoming_message.message_id
diff --git a/pkg/platform/sources/discord.py b/pkg/platform/sources/discord.py
index c279e714..da32c7ac 100644
--- a/pkg/platform/sources/discord.py
+++ b/pkg/platform/sources/discord.py
@@ -8,7 +8,6 @@ import base64
import uuid
import os
import datetime
-import io
import asyncio
from enum import Enum
diff --git a/pkg/platform/sources/qqbotpy.py b/pkg/platform/sources/qqbotpy.py
index 39c8dc8a..d4a4d526 100644
--- a/pkg/platform/sources/qqbotpy.py
+++ b/pkg/platform/sources/qqbotpy.py
@@ -501,7 +501,7 @@ class OfficialAdapter(adapter_model.MessagePlatformAdapter):
for event_handler in event_handler_mapping[event_type]:
setattr(self.bot, event_handler, wrapper)
except Exception as e:
- self.logger.error(f"Error in qqbotpy callback: {traceback.format_exc()}")
+ self.logger.error(f'Error in qqbotpy callback: {traceback.format_exc()}')
raise e
def unregister_listener(
diff --git a/pkg/platform/sources/telegram.py b/pkg/platform/sources/telegram.py
index 22ef63e8..d39bf23d 100644
--- a/pkg/platform/sources/telegram.py
+++ b/pkg/platform/sources/telegram.py
@@ -1,6 +1,5 @@
from __future__ import annotations
-import time
import telegram
import telegram.ext
diff --git a/pkg/platform/sources/webchat.py b/pkg/platform/sources/webchat.py
index f7f3d964..fce28bc2 100644
--- a/pkg/platform/sources/webchat.py
+++ b/pkg/platform/sources/webchat.py
@@ -133,7 +133,11 @@ class WebChatAdapter(msadapter.MessagePlatformAdapter):
)
# notify waiter
- session = (self.webchat_group_session if isinstance(message_source, platform_events.GroupMessage) else self.webchat_person_session)
+ session = (
+ self.webchat_group_session
+ if isinstance(message_source, platform_events.GroupMessage)
+ else self.webchat_person_session
+ )
if message_source.message_chain.message_id not in session.resp_waiters:
# session.resp_waiters[message_source.message_chain.message_id] = asyncio.Queue()
queue = session.resp_queues[message_source.message_chain.message_id]
@@ -147,10 +151,8 @@ class WebChatAdapter(msadapter.MessagePlatformAdapter):
# print(message_data)
await queue.put(message_data)
-
-
return message_data.model_dump()
-
+
async def is_stream_output_supported(self) -> bool:
return self.is_stream
@@ -186,7 +188,10 @@ class WebChatAdapter(msadapter.MessagePlatformAdapter):
await self.logger.info('WebChat调试适配器正在停止')
async def send_webchat_message(
- self, pipeline_uuid: str, session_type: str, message_chain_obj: typing.List[dict],
+ self,
+ pipeline_uuid: str,
+ session_type: str,
+ message_chain_obj: typing.List[dict],
is_stream: bool = False,
) -> dict:
self.is_stream = is_stream
@@ -202,7 +207,7 @@ class WebChatAdapter(msadapter.MessagePlatformAdapter):
if is_stream:
use_session.resp_queues[message_id] = asyncio.Queue()
- logger.debug(f"Initialized queue for message_id: {message_id}")
+ logger.debug(f'Initialized queue for message_id: {message_id}')
use_session.get_message_list(pipeline_uuid).append(
WebChatMessage(
diff --git a/pkg/platform/sources/wechatpad.py b/pkg/platform/sources/wechatpad.py
index 9bbb471d..895e77fb 100644
--- a/pkg/platform/sources/wechatpad.py
+++ b/pkg/platform/sources/wechatpad.py
@@ -241,8 +241,8 @@ class WeChatPadMessageConverter(adapter.MessageConverter):
# self.logger.info("_handler_compound_quote", ET.tostring(xml_data, encoding='unicode'))
appmsg_data = xml_data.find('.//appmsg')
quote_data = '' # 引用原文
- quote_id = None # 引用消息的原发送者
- tousername = None # 接收方: 所属微信的wxid
+ # quote_id = None # 引用消息的原发送者
+ # tousername = None # 接收方: 所属微信的wxid
user_data = '' # 用户消息
sender_id = xml_data.findtext('.//fromusername') # 发送方：单聊用户/群member
@@ -250,13 +250,10 @@ class WeChatPadMessageConverter(adapter.MessageConverter):
if appmsg_data:
user_data = appmsg_data.findtext('.//title') or ''
quote_data = appmsg_data.find('.//refermsg').findtext('.//content')
- quote_id = appmsg_data.find('.//refermsg').findtext('.//chatusr')
+ # quote_id = appmsg_data.find('.//refermsg').findtext('.//chatusr')
message_list.append(platform_message.WeChatAppMsg(app_msg=ET.tostring(appmsg_data, encoding='unicode')))
- if message:
- tousername = message['to_user_name']['str']
-
- _ = quote_id
- _ = tousername
+ # if message:
+ # tousername = message['to_user_name']['str']
if quote_data:
quote_data_message_list = platform_message.MessageChain()
diff --git a/pkg/platform/types/message.py b/pkg/platform/types/message.py
index 7dad4145..ecd7cc96 100644
--- a/pkg/platform/types/message.py
+++ b/pkg/platform/types/message.py
@@ -812,12 +812,14 @@ class File(MessageComponent):
def __str__(self):
return f'[文件]{self.name}'
+
class Face(MessageComponent):
"""系统表情
此处将超级表情骰子/划拳，一同归类于face
当face_type为rps(划拳)时 face_id 对应的是手势
当face_type为dice(骰子)时 face_id 对应的是点数
"""
+
type: str = 'Face'
"""表情类型"""
face_type: str = 'face'
@@ -834,15 +836,15 @@ class Face(MessageComponent):
elif self.face_type == 'rps':
return f'[表情]{self.face_name}({self.rps_data(self.face_id)})'
-
- def rps_data(self,face_id):
- rps_dict ={
- 1 : "布",
- 2 : "剪刀",
- 3 : "石头",
+ def rps_data(self, face_id):
+ rps_dict = {
+ 1: '布',
+ 2: '剪刀',
+ 3: '石头',
}
return rps_dict[face_id]
+
# ================ 个人微信专用组件 ================
@@ -971,5 +973,6 @@ class WeChatFile(MessageComponent):
"""文件地址"""
file_base64: str = ''
"""base64"""
+
def __str__(self):
- return f'[文件]{self.file_name}'
\ No newline at end of file
+ return f'[文件]{self.file_name}'
diff --git a/pkg/provider/entities.py b/pkg/provider/entities.py
index df2b5487..ff1e4526 100644
--- a/pkg/provider/entities.py
+++ b/pkg/provider/entities.py
@@ -127,6 +127,7 @@ class Message(pydantic.BaseModel):
class MessageChunk(pydantic.BaseModel):
"""消息"""
+
resp_message_id: typing.Optional[str] = None
"""消息id"""
@@ -148,7 +149,7 @@ class MessageChunk(pydantic.BaseModel):
tool_call_id: typing.Optional[str] = None
# tool_calls: typing.Optional[list[ToolCallChunk]] = None
-
+
is_final: bool = False
def readable_str(self) -> str:
@@ -210,6 +211,7 @@ class ToolCallChunk(pydantic.BaseModel):
function: FunctionCall
"""函数调用"""
+
class Prompt(pydantic.BaseModel):
"""供AI使用的Prompt"""
diff --git a/pkg/provider/modelmgr/requester.py b/pkg/provider/modelmgr/requester.py
index fa4a9ff8..d28783b9 100644
--- a/pkg/provider/modelmgr/requester.py
+++ b/pkg/provider/modelmgr/requester.py
@@ -94,19 +94,18 @@ class ProviderAPIRequester(metaclass=abc.ABCMeta):
extra_args (dict[str, typing.Any], optional): 额外的参数. Defaults to {}.
Returns:
- llm_entities.Message | typing.AsyncGenerator[llm_entities.MessageChunk]: 返回消息对象
+ llm_entities.Message: 返回消息对象
"""
pass
@abc.abstractmethod
async def invoke_llm_stream(
- self,
- query: core_entities.Query,
- model: RuntimeLLMModel,
- messages: typing.List[llm_entities.Message],
- funcs: typing.List[tools_entities.LLMFunction] = None,
- stream: bool = False,
- extra_args: dict[str, typing.Any] = {},
+ self,
+ query: core_entities.Query,
+ model: RuntimeLLMModel,
+ messages: typing.List[llm_entities.Message],
+ funcs: typing.List[tools_entities.LLMFunction] = None,
+ extra_args: dict[str, typing.Any] = {},
) -> llm_entities.MessageChunk:
"""调用API
@@ -117,7 +116,7 @@ class ProviderAPIRequester(metaclass=abc.ABCMeta):
extra_args (dict[str, typing.Any], optional): 额外的参数. Defaults to {}.
Returns:
- llm_entities.Message | typing.AsyncGenerator[llm_entities.MessageChunk]: 返回消息对象
+ typing.AsyncGenerator[llm_entities.MessageChunk]: 返回消息对象
"""
pass
diff --git a/pkg/provider/modelmgr/requesters/chatcmpl.py b/pkg/provider/modelmgr/requesters/chatcmpl.py
index d5c3b90a..4fcce481 100644
--- a/pkg/provider/modelmgr/requesters/chatcmpl.py
+++ b/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, app
+from ....core import entities as core_entities
from ... import entities as llm_entities
from ...tools import entities as tools_entities
@@ -129,12 +129,10 @@ class OpenAIChatCompletions(requester.ProviderAPIRequester):
req_messages: list[dict],
use_model: requester.RuntimeLLMModel,
use_funcs: list[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
- ) ->llm_entities.MessageChunk:
+ ) -> llm_entities.MessageChunk:
self.client.api_key = use_model.token_mgr.get_token()
-
args = {}
args['model'] = use_model.model_entity.name
@@ -158,43 +156,42 @@ class OpenAIChatCompletions(requester.ProviderAPIRequester):
args['messages'] = messages
- if stream:
- current_content = ''
- 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(pipeline_config, chunk, chunk_idx)
- if delta_message.content:
- current_content += delta_message.content
- delta_message.content = 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
+ current_content = ''
+ 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(pipeline_config, chunk, chunk_idx)
+ if delta_message.content:
+ current_content += delta_message.content
+ delta_message.content = 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
+ 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
- if chunk_idx % 64 == 0 or delta_message.is_final:
- yield delta_message
- # return
+ if chunk_idx % 64 == 0 or delta_message.is_final:
+ yield delta_message
+ # return
async def _closure(
self,
@@ -202,7 +199,6 @@ class OpenAIChatCompletions(requester.ProviderAPIRequester):
req_messages: list[dict],
use_model: requester.RuntimeLLMModel,
use_funcs: list[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
) -> llm_entities.Message:
self.client.api_key = use_model.token_mgr.get_token()
@@ -317,7 +313,6 @@ class OpenAIChatCompletions(requester.ProviderAPIRequester):
model: requester.RuntimeLLMModel,
messages: typing.List[llm_entities.Message],
funcs: typing.List[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
) -> llm_entities.MessageChunk:
req_messages = [] # req_messages 仅用于类内，外部同步由 query.messages 进行
@@ -337,7 +332,6 @@ class OpenAIChatCompletions(requester.ProviderAPIRequester):
req_messages=req_messages,
use_model=model,
use_funcs=funcs,
- stream=stream,
extra_args=extra_args,
):
yield item
diff --git a/pkg/provider/modelmgr/requesters/giteeaichatcmpl.py b/pkg/provider/modelmgr/requesters/giteeaichatcmpl.py
index 2a618c9f..1c19a534 100644
--- a/pkg/provider/modelmgr/requesters/giteeaichatcmpl.py
+++ b/pkg/provider/modelmgr/requesters/giteeaichatcmpl.py
@@ -12,7 +12,6 @@ import re
import openai.types.chat.chat_completion as chat_completion
-
class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
"""Gitee AI ChatCompletions API 请求器"""
@@ -20,7 +19,7 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
'base_url': 'https://ai.gitee.com/v1',
'timeout': 120,
}
- is_think:bool = False
+ is_think: bool = False
async def _closure(
self,
@@ -52,15 +51,14 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
pipeline_config = query.pipeline_config
- message = await self._make_msg(resp,pipeline_config)
+ message = await self._make_msg(resp, pipeline_config)
return message
-
async def _make_msg(
- self,
- chat_completion: chat_completion.ChatCompletion,
- pipeline_config: dict[str, typing.Any] = {'trigger': {'misc': {'remove_think': False}}},
+ self,
+ chat_completion: chat_completion.ChatCompletion,
+ pipeline_config: dict[str, typing.Any] = {'trigger': {'misc': {'remove_think': False}}},
) -> llm_entities.Message:
chatcmpl_message = chat_completion.choices[0].message.model_dump()
# print(chatcmpl_message.keys(), chatcmpl_message.values())
@@ -73,23 +71,25 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
# deepseek的reasoner模型
if pipeline_config['trigger'].get('misc', '').get('remove_think'):
- chatcmpl_message['content'] = re.sub(r'.*?', '', chatcmpl_message['content'], flags=re.DOTALL)
+ chatcmpl_message['content'] = re.sub(
+ r'.*?', '', chatcmpl_message['content'], flags=re.DOTALL
+ )
else:
if reasoning_content is not None:
- chatcmpl_message['content'] = '\n' + reasoning_content + '\n\n' + chatcmpl_message['content']
+ chatcmpl_message['content'] = (
+ '\n' + reasoning_content + '\n\n' + chatcmpl_message['content']
+ )
message = llm_entities.Message(**chatcmpl_message)
return message
-
async def _make_msg_chunk(
self,
pipeline_config: dict[str, typing.Any],
chat_completion: chat_completion.ChatCompletion,
idx: int,
) -> llm_entities.MessageChunk:
-
# 处理流式chunk和完整响应的差异
# print(chat_completion.choices[0])
if hasattr(chat_completion, 'choices'):
@@ -104,7 +104,6 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
if 'role' not in delta or delta['role'] is None:
delta['role'] = 'assistant'
-
reasoning_content = delta['reasoning_content'] if 'reasoning_content' in delta else None
delta['content'] = '' if delta['content'] is None else delta['content']
@@ -115,7 +114,7 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
if delta['content'] == '':
self.is_think = True
delta['content'] = ''
- if delta['content'] == rf'':
+ if delta['content'] == r'':
self.is_think = False
delta['content'] = ''
if not self.is_think:
@@ -126,7 +125,6 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
if reasoning_content is not None:
delta['content'] += reasoning_content
-
message = llm_entities.MessageChunk(**delta)
return message
@@ -137,7 +135,6 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
req_messages: list[dict],
use_model: requester.RuntimeLLMModel,
use_funcs: list[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
) -> llm_entities.Message | typing.AsyncGenerator[llm_entities.MessageChunk, None]:
self.client.api_key = use_model.token_mgr.get_token()
@@ -165,44 +162,38 @@ class GiteeAIChatCompletions(chatcmpl.OpenAIChatCompletions):
args['messages'] = messages
- if stream:
- current_content = ''
- 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(pipeline_config,chunk,chunk_idx)
- if delta_message.content:
- current_content += delta_message.content
- delta_message.content = 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
-
- if chunk_idx % 64 == 0 or delta_message.is_final:
-
- yield delta_message
+ current_content = ''
+ 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(pipeline_config, chunk, chunk_idx)
+ if delta_message.content:
+ current_content += delta_message.content
+ delta_message.content = 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
+ if chunk_idx % 64 == 0 or delta_message.is_final:
+ yield delta_message
diff --git a/pkg/provider/modelmgr/requesters/modelscopechatcmpl.py b/pkg/provider/modelmgr/requesters/modelscopechatcmpl.py
index c1888a5e..97201e47 100644
--- a/pkg/provider/modelmgr/requesters/modelscopechatcmpl.py
+++ b/pkg/provider/modelmgr/requesters/modelscopechatcmpl.py
@@ -165,11 +165,10 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
return message
async def _req_stream(
- self,
- args: dict,
- extra_body: dict = {},
+ self,
+ args: dict,
+ extra_body: dict = {},
) -> chat_completion.ChatCompletion:
-
async for chunk in await self.client.chat.completions.create(**args, extra_body=extra_body):
yield chunk
@@ -179,7 +178,6 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
chat_completion: chat_completion.ChatCompletion,
idx: int,
) -> llm_entities.MessageChunk:
-
# 处理流式chunk和完整响应的差异
# print(chat_completion.choices[0])
if hasattr(chat_completion, 'choices'):
@@ -195,7 +193,6 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
if 'role' not in delta or delta['role'] is None:
delta['role'] = 'assistant'
-
reasoning_content = delta['reasoning_content'] if 'reasoning_content' in delta else None
delta['content'] = '' if delta['content'] is None else delta['content']
@@ -203,13 +200,13 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
# deepseek的reasoner模型
if pipeline_config['trigger'].get('misc', '').get('remove_think'):
- if reasoning_content is not None :
+ if reasoning_content is not None:
pass
else:
delta['content'] = delta['content']
else:
if reasoning_content is not None and idx == 0:
- delta['content'] += f'\n{reasoning_content}'
+ delta['content'] += f'\n{reasoning_content}'
elif reasoning_content is None:
if self.is_content:
delta['content'] = delta['content']
@@ -219,7 +216,6 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
else:
delta['content'] += reasoning_content
-
message = llm_entities.MessageChunk(**delta)
return message
@@ -230,7 +226,6 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
req_messages: list[dict],
use_model: requester.RuntimeLLMModel,
use_funcs: list[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
) -> llm_entities.Message | typing.AsyncGenerator[llm_entities.MessageChunk, None]:
self.client.api_key = use_model.token_mgr.get_token()
@@ -258,48 +253,42 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
args['messages'] = messages
- if stream:
- current_content = ''
- 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(pipeline_config,chunk,chunk_idx)
- if delta_message.content:
- current_content += delta_message.content
- delta_message.content = 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
-
- if chunk_idx % 64 == 0 or delta_message.is_final:
-
- yield delta_message
- # return
+ current_content = ''
+ 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(pipeline_config, chunk, chunk_idx)
+ if delta_message.content:
+ current_content += delta_message.content
+ delta_message.content = 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
+ if chunk_idx % 64 == 0 or delta_message.is_final:
+ yield delta_message
+ # return
async def invoke_llm(
self,
@@ -340,16 +329,14 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
except openai.APIError as e:
raise errors.RequesterError(f'请求错误: {e.message}')
-
async def invoke_llm_stream(
self,
query: core_entities.Query,
model: requester.RuntimeLLMModel,
messages: typing.List[llm_entities.Message],
funcs: typing.List[tools_entities.LLMFunction] = None,
- stream: bool = False,
extra_args: dict[str, typing.Any] = {},
- ) -> llm_entities.MessageChunk:
+ ) -> llm_entities.MessageChunk:
req_messages = [] # req_messages 仅用于类内，外部同步由 query.messages 进行
for m in messages:
msg_dict = m.dict(exclude_none=True)
@@ -367,7 +354,6 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
req_messages=req_messages,
use_model=model,
use_funcs=funcs,
- stream=stream,
extra_args=extra_args,
):
yield item
@@ -386,4 +372,4 @@ class ModelScopeChatCompletions(requester.ProviderAPIRequester):
except openai.RateLimitError as e:
raise errors.RequesterError(f'请求过于频繁或余额不足: {e.message}')
except openai.APIError as e:
- raise errors.RequesterError(f'请求错误: {e.message}')
\ No newline at end of file
+ raise errors.RequesterError(f'请求错误: {e.message}')
diff --git a/pkg/provider/modelmgr/requesters/ppiochatcmpl.py b/pkg/provider/modelmgr/requesters/ppiochatcmpl.py
index 85b321a7..46da6e01 100644
--- a/pkg/provider/modelmgr/requesters/ppiochatcmpl.py
+++ b/pkg/provider/modelmgr/requesters/ppiochatcmpl.py
@@ -5,8 +5,8 @@ import typing
from . import chatcmpl
import openai.types.chat.chat_completion as chat_completion
-from .. import errors, requester
-from ....core import entities as core_entities, app
+from .. import requester
+from ....core import entities as core_entities
from ... import entities as llm_entities
from ...tools import entities as tools_entities
import re
@@ -25,9 +25,9 @@ class PPIOChatCompletions(chatcmpl.OpenAIChatCompletions):
is_think: bool = False
async def _make_msg(
- self,
- chat_completion: chat_completion.ChatCompletion,
- pipeline_config: dict[str, typing.Any] = {'trigger': {'misc': {'remove_think': False}}},
+ self,
+ chat_completion: chat_completion.ChatCompletion,
+ pipeline_config: dict[str, typing.Any] = {'trigger': {'misc': {'remove_think': False}}},
) -> llm_entities.Message:
chatcmpl_message = chat_completion.choices[0].message.model_dump()
# print(chatcmpl_message.keys(), chatcmpl_message.values())
@@ -40,21 +40,24 @@ class PPIOChatCompletions(chatcmpl.OpenAIChatCompletions):
# deepseek的reasoner模型
if pipeline_config['trigger'].get('misc', '').get('remove_think'):
- chatcmpl_message['content'] = re.sub(r'.*?', '', chatcmpl_message['content'], flags=re.DOTALL)
+ chatcmpl_message['content'] = re.sub(
+ r'.*?', '', chatcmpl_message['content'], flags=re.DOTALL
+ )
else:
if reasoning_content is not None:
- chatcmpl_message['content'] = '\n' + reasoning_content + '\n\n' + chatcmpl_message['content']
+ chatcmpl_message['content'] = (
+ '\n' + reasoning_content + '\n\n' + chatcmpl_message['content']
+ )
message = llm_entities.Message(**chatcmpl_message)
return message
-
async def _make_msg_chunk(
- self,
- pipeline_config: dict[str, typing.Any],
- chat_completion: chat_completion.ChatCompletion,
- idx: int,
+ self,
+ pipeline_config: dict[str, typing.Any],
+ chat_completion: chat_completion.ChatCompletion,
+ idx: int,
) -> llm_entities.MessageChunk:
# 处理流式chunk和完整响应的差异
# print(chat_completion.choices[0])
@@ -80,7 +83,7 @@ class PPIOChatCompletions(chatcmpl.OpenAIChatCompletions):
if '' in delta['content']:
self.is_think = True
delta['content'] = ''
- if rf'' in delta['content']:
+ if r'' in delta['content']:
self.is_think = False
delta['content'] = ''
if not self.is_think:
@@ -95,15 +98,13 @@ class PPIOChatCompletions(chatcmpl.OpenAIChatCompletions):
return message
-
async def _closure_stream(
- self,
- query: core_entities.Query,
- req_messages: list[dict],
- use_model: requester.RuntimeLLMModel,
- use_funcs: list[tools_entities.LLMFunction] = None,
- stream: bool = False,
- extra_args: dict[str, typing.Any] = {},
+ self,
+ query: core_entities.Query,
+ req_messages: list[dict],
+ use_model: requester.RuntimeLLMModel,
+ use_funcs: list[tools_entities.LLMFunction] = None,
+ extra_args: dict[str, typing.Any] = {},
) -> llm_entities.Message | typing.AsyncGenerator[llm_entities.MessageChunk, None]:
self.client.api_key = use_model.token_mgr.get_token()
@@ -130,40 +131,38 @@ class PPIOChatCompletions(chatcmpl.OpenAIChatCompletions):
args['messages'] = messages
- if stream:
- current_content = ''
- 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(pipeline_config, chunk, chunk_idx)
- if delta_message.content:
- current_content += delta_message.content
- delta_message.content = 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
+ current_content = ''
+ 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(pipeline_config, chunk, chunk_idx)
+ if delta_message.content:
+ current_content += delta_message.content
+ delta_message.content = 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
+ 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
- if chunk_idx % 64 == 0 or delta_message.is_final:
- yield delta_message
+ if chunk_idx % 64 == 0 or delta_message.is_final:
+ yield delta_message
diff --git a/pkg/provider/runners/difysvapi.py b/pkg/provider/runners/difysvapi.py
index 8182cc54..40a3140c 100644
--- a/pkg/provider/runners/difysvapi.py
+++ b/pkg/provider/runners/difysvapi.py
@@ -348,7 +348,9 @@ class DifyServiceAPIRunner(runner.RequestRunner):
except AttributeError:
is_stream = False
- batch_pending_index = 0
+ _ = is_stream
+
+ # batch_pending_index = 0
plain_text, image_ids = await self._preprocess_user_message(query)
diff --git a/pkg/provider/runners/localagent.py b/pkg/provider/runners/localagent.py
index 599b0b08..3ff0ce9d 100644
--- a/pkg/provider/runners/localagent.py
+++ b/pkg/provider/runners/localagent.py
@@ -128,8 +128,7 @@ class LocalAgentRunner(runner.RequestRunner):
id=tool_call.id,
type=tool_call.type,
function=llm_entities.FunctionCall(
- name=tool_call.function.name if tool_call.function else '',
- arguments=''
+ name=tool_call.function.name if tool_call.function else '', arguments=''
),
)
if tool_call.function and tool_call.function.arguments:
diff --git a/pkg/utils/image.py b/pkg/utils/image.py
index f69d29d2..d9518e12 100644
--- a/pkg/utils/image.py
+++ b/pkg/utils/image.py
@@ -204,9 +204,9 @@ async def get_slack_image_to_base64(pic_url: str, bot_token: str):
try:
async with aiohttp.ClientSession() as session:
async with session.get(pic_url, headers=headers) as resp:
- mime_type = resp.headers.get("Content-Type", "application/octet-stream")
+ mime_type = resp.headers.get('Content-Type', 'application/octet-stream')
file_bytes = await resp.read()
- base64_str = base64.b64encode(file_bytes).decode("utf-8")
- return f"data:{mime_type};base64,{base64_str}"
+ base64_str = base64.b64encode(file_bytes).decode('utf-8')
+ return f'data:{mime_type};base64,{base64_str}'
except Exception as e:
- raise (e)
\ No newline at end of file
+ raise (e)
diff --git a/pkg/utils/importutil.py b/pkg/utils/importutil.py
index 8acc5c45..1933d611 100644
--- a/pkg/utils/importutil.py
+++ b/pkg/utils/importutil.py
@@ -32,7 +32,7 @@ def import_dir(path: str):
rel_path = full_path.replace(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), '')
rel_path = rel_path[1:]
rel_path = rel_path.replace('/', '.')[:-3]
- rel_path = rel_path.replace("\\",".")
+ rel_path = rel_path.replace('\\', '.')
importlib.import_module(rel_path)