流式基本流程已通过修改了yield和return的冲突导致的问题

pkg/pipeline/process/handlers/chat.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from itertools import accumulate
+import uuid
 import typing
 import traceback
 
@@ -59,8 +59,11 @@ class ChatMessageHandler(handler.MessageHandler):
                 query.user_message.content = event_ctx.event.alter
 
             text_length = 0
-
-            is_stream = query.adapter.is_stream_output_supported()
+            try:
+                is_stream = query.adapter.is_stream
+            except AttributeError:
+                is_stream = False
+            print(is_stream)
 
             try:
                 for r in runner_module.preregistered_runners:
@@ -68,23 +71,28 @@ class ChatMessageHandler(handler.MessageHandler):
                         runner = r(self.ap, query.pipeline_config)
                         break
                 else:
-                    raise ValueError(f'Request runner not found: {query.pipeline_config["ai"]["runner"]["runner"]}')
-
+                    raise ValueError(f'未找到请求运行器: {query.pipeline_config["ai"]["runner"]["runner"]}')
                 if is_stream:
-                    async for results in runner.run(query):
-                        async for result in results:
+                    resp_message_id = uuid.uuid4()
+                    await query.adapter.create_message_card(resp_message_id, query.message_event)
+                    async for result in runner.run(query):
+                        result.resp_message_id = resp_message_id
+                        if query.resp_messages:
+                            query.resp_messages.pop()
+                        if query.resp_message_chain:
+                            query.resp_message_chain.pop()
 
-                            query.resp_messages.append(result)
+                        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)
 
-                            if result.content is not None:
-                                text_length += len(result.content)
-                        
-                            yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+                        yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+                    # else:
+                    #     yield entities.StageProcessResult(result_type=entities.ResultType.INTERRUPT, new_query=query)
 
                 else:
-
                     async for result in runner.run(query):
                         query.resp_messages.append(result)
 
@@ -96,7 +104,7 @@ class ChatMessageHandler(handler.MessageHandler):
                         yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
 
                 query.session.using_conversation.messages.append(query.user_message)
-                
+
                 query.session.using_conversation.messages.extend(query.resp_messages)
             except Exception as e:
                 self.ap.logger.error(f'Request failed({query.query_id}): {type(e).__name__} {str(e)}')

pkg/pipeline/respback/respback.py

@@ -3,6 +3,7 @@ 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
@@ -39,12 +40,16 @@ class SendResponseBackStage(stage.PipelineStage):
         quote_origin = query.pipeline_config['output']['misc']['quote-origin']
 
         has_chunks = any(isinstance(msg, llm_entities.MessageChunk) for msg in query.resp_messages)
+        print(has_chunks)
         if has_chunks and hasattr(query.adapter,'reply_message_chunk'):
+            is_final = [msg.is_final for msg in query.resp_messages][0]
+            print(is_final)
             await query.adapter.reply_message_chunk(
                 message_source=query.message_event,
-                message_id=query.query_id,
-                message_generator=query.resp_message_chain[-1],
+                message_id=query.message_event.message_chain.message_id,
+                message=query.resp_message_chain[-1],
                 quote_origin=quote_origin,
+                is_final=is_final,
             )
         else:
             await query.adapter.reply_message(