mirror of
https://github.com/langgenius/dify.git
synced 2026-04-03 09:00:25 -04:00
24 lines
940 B
Python
24 lines
940 B
Python
from collections.abc import Generator, Mapping
|
|
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class BaseBackwardsInvocation:
|
|
@classmethod
|
|
def convert_to_event_stream(cls, response: Generator[BaseModel | Mapping | str, None, None] | BaseModel | Mapping):
|
|
if isinstance(response, Generator):
|
|
try:
|
|
for chunk in response:
|
|
if isinstance(chunk, BaseModel | dict):
|
|
yield BaseBackwardsInvocationResponse(data=chunk).model_dump_json().encode()
|
|
except Exception as e:
|
|
error_message = BaseBackwardsInvocationResponse(error=str(e)).model_dump_json()
|
|
yield error_message.encode()
|
|
else:
|
|
yield BaseBackwardsInvocationResponse(data=response).model_dump_json().encode()
|
|
|
|
|
|
class BaseBackwardsInvocationResponse[T: dict | Mapping | str | bool | int | BaseModel](BaseModel):
|
|
data: T | None = None
|
|
error: str = ""
|