Why the new version (v1.0.5) abandoned the celery and socketio technologies?

[wfg314]

In the older version v0.6.19, celery and socketio were used, but in the new version v1.0.5, the socketio service is commented out, and celery is only defined but not used。why?

v0.6.19:

v1.0.5

[dosubot[bot]]

Hey @wfg314, I'm here to help you with any issues or questions you have about the project. Let's tackle this together!

The new version (v1.0.5) of Langflow abandoned the use of Celery and Socket.IO technologies due to specific reasons:

1. Celery: The decision to abandon Celery was due to its unavailability or import issues. The USE_CELERY flag is set to False when an ImportError occurs, indicating that Celery could not be imported or was not available. The code falls back to using the AnyIOBackend instead of the CeleryBackend ^[1].

2. Socket.IO: The new version replaced Socket.IO with WebSocket for real-time communication or event handling. The WebSocket implementation is used to establish connections, handle messages, and manage errors and reconnections ^[2].

These changes were likely made to simplify the setup and reduce dependencies, as AnyIOBackend and WebSocket are more lightweight and easier to maintain compared to Celery and Socket.IO.

To continue talking to Dosu, mention @dosu.

[wfg314]

but your snippet can only found in v0.6.19, not v1.0.5 or dev. I mean how to use in version v1.0.5 or dev ?

[dosubot[bot]]

In the new version (v1.0.5) or the dev branch of the Langflow project, WebSocket connections are managed using the ChatService class, which handles the real-time communication through WebSocket endpoints. Here is a summary of how WebSocket connections are established and managed:

1. WebSocket Endpoint:
   The WebSocket connection is established through the /chat/{client_id} endpoint defined in the chat.py file:

   @router.websocket("/chat/{client_id}")
   async def chat(
       client_id: str,
       websocket: WebSocket,
       db: Session = Depends(get_session),
       chat_service: "ChatService" = Depends(get_chat_service),
   ):
       """Websocket endpoint for chat."""
       try:
           user = await get_current_user_for_websocket(websocket, db)
           await websocket.accept()
           if not user:
               await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Unauthorized")
           elif not user.is_active:
               await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Unauthorized")
           if client_id in chat_service.cache_service:
               await chat_service.handle_websocket(client_id, websocket)
           else:
               message = "Please, build the flow before sending messages"
               await websocket.close(code=status.WS_1011_INTERNAL_ERROR, reason=message)
       except WebSocketException as exc:
           logger.error(f"Websocket error: {exc}")
           await websocket.close(code=status.WS_1011_INTERNAL_ERROR, reason=str(exc))
       except Exception as exc:
           logger.error(f"Error in chat websocket: {exc}")
           message = exc.detail if isinstance(exc, HTTPException) else str(exc)
           if "Could not validate credentials" in str(exc):
               await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Unauthorized")
           else:
               await websocket.close(code=status.WS_1011_INTERNAL_ERROR, reason=message)

2. Handling WebSocket Connections:
   The ChatService class is responsible for managing the WebSocket connections and handling the messages. The handle_websocket method in the ChatService class processes the incoming messages and sends responses back to the client.

3. Real-Time Communication:
   The real-time communication is handled by sending JSON responses through the WebSocket connection. The ChatService class uses the WebSocket connection to send various types of messages based on the state of the chat or the flow being executed.

This code shows how WebSocket connections are established, managed, and used for real-time communication in the newer versions of the Langflow project. The ChatService class is responsible for handling the WebSocket connections and sending messages to the client in real-time ^[1][2].

To continue talking to Dosu, mention @dosu.

[wfg314]

No,your answer also base on v0.6.19.

[dosubot[bot]]

In the new version (v1.0.5) or the dev branch of the Langflow project, WebSocket connections are established and managed using the socketio library. The backend utilizes WebSocket in the following ways:

1. Setting Up the SocketIO Server:

   def set_socketio_server(socketio_server):
       from langflow.services.deps import get_socket_service
   
       socket_service = get_socket_service()
       socket_service.init(socketio_server)

2. Handling WebSocket Events:

   async def get_vertices(sio, sid, flow_id, chat_service):
       try:
           session = get_session()
           flow: Flow = session.exec(select(Flow).where(Flow.id == flow_id)).first()
           if not flow or not flow.data:
               await sio.emit("error", data="Invalid flow ID", to=sid)
               return
   
           graph = Graph.from_payload(flow.data)
           chat_service.set_cache(flow_id, graph)
           vertices = graph.layered_topological_sort()
   
           # Emit the vertices to the client
           await sio.emit("vertices_order", data=vertices, to=sid)
   
       except Exception as exc:
           await sio.emit("error", data=str(exc), to=sid)

3. Building a Vertex:

   async def build_vertex(
       sio: socketio.AsyncServer,
       sid: str,
       flow_id: str,
       vertex_id: str,
       get_cache: Callable,
       set_cache: Callable,
       tweaks=None,
       inputs=None,
   ):
       try:
           cache = get_cache(flow_id)
           graph = cache.get("result")
   
           if not isinstance(graph, Graph):
               await sio.emit("error", data="Invalid graph", to=sid)
               return
           vertex = graph.get_vertex(vertex_id)
           if not vertex:
               await sio.emit("error", data="Invalid vertex", to=sid)
               return
           start_time = time.perf_counter()
           try:
               if isinstance(vertex, Vertex) or not vertex._built:
                   await vertex.build(user_id=None, session_id=sid)
               params = vertex._built_object_repr()
               valid = True
               result_dict = vertex.get_built_result()
               vertex.set_artifacts()
               artifacts = vertex.artifacts
               timedelta = time.perf_counter() - start_time
               duration = format_elapsed_time(timedelta)
               result_dict = ResultDataResponse(
                   results=result_dict,
                   artifacts=artifacts,
                   duration=duration,
                   timedelta=timedelta,
               )
           except Exception as exc:
               params = str(exc)
               valid = False
               result_dict = ResultDataResponse(results={})
               artifacts = {}
           set_cache(flow_id, graph)
           await log_vertex_build(
               flow_id=flow_id,
               vertex_id=vertex_id,
               valid=valid,
               params=params,
               data=result_dict,
               artifacts=artifacts,
           )
   
           # Emit the vertex build response
           response = VertexBuildResponse(valid=valid, params=params, id=vertex.id, data=result_dict)
           await sio.emit("vertex_build", data=response.model_dump(), to=sid)
   
       except Exception as exc:
           await sio.emit("error", data=str(exc), to=sid)

Additionally, the AsyncStreamingLLMCallbackHandler class in callback.py is used to handle streaming responses from the LLM and send them via WebSocket:

class AsyncStreamingLLMCallbackHandler(AsyncCallbackHandler):
    """Callback handler for streaming LLM responses."""

    def __init__(self, client_id: str):
        self.chat_service = get_chat_service()
        self.client_id = client_id
        self.websocket = self.chat_service.active_connections[self.client_id]

    async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        resp = ChatResponse(message=token, type="stream", intermediate_steps="")
        await self.websocket.send_json(resp.model_dump())

    async def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> Any:
        resp = ChatResponse(
            message="",
            type="stream",
            intermediate_steps=f"Tool input: {input_str}",
        )
        await self.websocket.send_json(resp.model_dump())

    async def on_tool_end(self, output: str, **kwargs: Any) -> Any:
        observation_prefix = kwargs.get("observation_prefix", "Tool output: ")
        split_output = output.split()
        first_word = split_output[0]
        rest_of_output = split_output[1:]
        intermediate_steps = f"{observation_prefix}{first_word}"

        resp = ChatResponse(
            message="",
            type="stream",
            intermediate_steps=intermediate_steps,
        )
        rest_of_resps = [
            ChatResponse(
                message="",
                type="stream",
                intermediate_steps=f"{word}",
            )
            for word in rest_of_output
        ]
        resps = [resp] + rest_of_resps
        try:
            for resp in resps:
                await self.websocket.send_json(resp.model_dump())
        except Exception as exc:
            logger.error(f"Error sending response: {exc}")

    async def on_tool_error(self, error: BaseException, **kwargs: Any) -> None:
        pass

    async def on_text(self, text: str, **kwargs: Any) -> Any:
        if "Prompt after formatting" in text:
            text = text.replace("Prompt after formatting:\n", "")
            text = remove_ansi_escape_codes(text)
            resp = PromptResponse(prompt=text)
            await self.websocket.send_json(resp.model_dump())
            self.chat_service.chat_history.add_message(self.client_id, resp)

    async def on_agent_action(self, action: AgentAction, **kwargs: Any):
        log = f"Thought: {action.log}"
        if "\n" in log:
            logs = log.split("\n")
            for log in logs:
                resp = ChatResponse(message="", type="stream", intermediate_steps=log)
                await self.websocket.send_json(resp.model_dump())
        else:
            resp = ChatResponse(message="", type="stream", intermediate_steps=log)
            await self.websocket.send_json(resp.model_dump())

    async def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
        resp = ChatResponse(
            message="",
            type="stream",
            intermediate_steps=finish.log,
        )
        await self.websocket.send_json(resp.model_dump())

These methods and classes collectively manage the WebSocket connections, including setting up the server, handling incoming events, processing vertex builds, and streaming responses to the client ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[wfg314]

no,your answer still based on v0.6.19, not v1.0.5

[Cristhianzl]

Hi! 👋

We are using other channels as our official means of communication with users. We apologize for the delayed response.
For faster assistance, please direct your questions to our Discord channel or open a GitHub issue, as we are much more active on these platforms.

Langflow Discord Server

Thank you for your understanding.

Best regards,
The Langflow Team