xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-07-20 03:16:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: ruff format & remove stream params in requester

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