diff --git a/.vscode/launch.json b/.vscode/launch.json index a8229b155..3332b67e9 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -17,6 +17,9 @@ ], "jinja": true, "justMyCode": true, + "env": { + "LANGFLOW_LOG_LEVEL": "debug" + }, "envFile": "${workspaceFolder}/.env" }, { diff --git a/src/backend/langflow/api/v1/chat.py b/src/backend/langflow/api/v1/chat.py index 7f85bb211..e1b52aeb5 100644 --- a/src/backend/langflow/api/v1/chat.py +++ b/src/backend/langflow/api/v1/chat.py @@ -1,6 +1,6 @@ import time -from typing import Optional import uuid +from typing import TYPE_CHECKING, Optional from fastapi import ( APIRouter, @@ -35,6 +35,9 @@ from langflow.services.chat.service import ChatService from langflow.services.deps import get_chat_service, get_session from langflow.services.monitor.utils import log_vertex_build +if TYPE_CHECKING: + from langflow.graph.vertex.types import ChatVertex + router = APIRouter(tags=["Chat"]) @@ -237,27 +240,40 @@ async def build_vertex_stream( else: graph = cache.get("result") - vertex = graph.get_vertex(vertex_id) + vertex: "ChatVertex" = graph.get_vertex(vertex_id) + if not hasattr(vertex, "stream"): + raise ValueError(f"Vertex {vertex_id} does not support streaming") if not vertex.pinned or not vertex._built: + logger.debug(f"Streaming vertex {vertex_id}") stream_data = StreamData( event="message", - data={"message": "Building vertex"}, + data={"message": f"Streaming vertex {vertex_id}"}, ) yield str(stream_data) - + number_of_chunks = 0 async for chunk in vertex.stream(): stream_data = StreamData( event="message", data={"chunk": chunk}, ) + number_of_chunks += 1 yield str(stream_data) + logger.debug(f"Number of chunks: {number_of_chunks}") + elif vertex.result is not None: + stream_data = StreamData( + event="message", + data={"chunk": vertex._built_result}, + ) + yield str(stream_data) else: raise ValueError(f"No result found for vertex {vertex_id}") except Exception as exc: + logger.error(f"Error building vertex: {exc}") yield str(StreamData(event="error", data={"error": str(exc)})) - - yield str(StreamData(event="close", data={"message": "Stream closed"})) + finally: + logger.debug("Closing stream") + yield str(StreamData(event="close", data={"message": "Stream closed"})) return StreamingResponse(stream_vertex(), media_type="text/event-stream") except Exception as exc: diff --git a/src/backend/langflow/api/v1/endpoints.py b/src/backend/langflow/api/v1/endpoints.py index 2dc79e85a..fce4f9649 100644 --- a/src/backend/langflow/api/v1/endpoints.py +++ b/src/backend/langflow/api/v1/endpoints.py @@ -220,7 +220,9 @@ async def preload_flow( raise HTTPException(status_code=500, detail=str(exc)) from exc -@router.post("/run/{flow_id}", response_model=ProcessResponse) +@router.post( + "/run/{flow_id}", response_model=RunResponse, response_model_exclude_none=True +) async def run_flow_with_caching( session: Annotated[Session, Depends(get_session)], flow_id: str, @@ -235,13 +237,13 @@ async def run_flow_with_caching( session_data = await session_service.load_session(session_id) graph, artifacts = session_data if session_data else (None, None) task_result: Any = None - task_status = None if not graph: raise ValueError("Graph not found in the session") task_result = await run_graph( - graph, - session_id, - inputs, + graph=graph, + flow_id=flow_id, + session_id=session_id, + inputs=inputs, artifacts=artifacts, session_service=session_service, ) @@ -262,16 +264,15 @@ async def run_flow_with_caching( graph_data = flow.data graph_data = process_tweaks(graph_data, tweaks) task_result = await run_graph( - graph_data, - inputs, - tweaks, - session_id, + graph=graph_data, + flow_id=flow_id, + session_id=session_id, + inputs=inputs, + artifacts={}, session_service=session_service, ) - return RunResponse( - outputs=task_result, session_id=session_id, status=task_status - ) + return RunResponse(outputs=task_result, session_id=session_id) except sa.exc.StatementError as exc: # StatementError('(builtins.ValueError) badly formed hexadecimal UUID string') if "badly formed hexadecimal UUID string" in str(exc): diff --git a/src/backend/langflow/api/v1/schemas.py b/src/backend/langflow/api/v1/schemas.py index 0c33fdb36..0092efa4e 100644 --- a/src/backend/langflow/api/v1/schemas.py +++ b/src/backend/langflow/api/v1/schemas.py @@ -4,7 +4,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Union from uuid import UUID -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_serializer from langflow.services.database.models.api_key.model import ApiKeyRead from langflow.services.database.models.base import orjson_dumps @@ -70,9 +70,21 @@ class RunResponse(BaseModel): """Run response schema.""" outputs: Optional[List[Any]] = None - status: Optional[str] = None session_id: Optional[str] = None + @model_serializer(mode="wrap") + def serialize(self, handler): + # Serialize all the outputs if they are base models + if self.outputs: + serialized_outputs = [] + for output in self.outputs: + if isinstance(output, BaseModel): + serialized_outputs.append(output.model_dump(exclude_none=True)) + else: + serialized_outputs.append(output) + self.outputs = serialized_outputs + return handler(self) + class PreloadResponse(BaseModel): """Preload response schema.""" diff --git a/src/backend/langflow/components/chains/ConversationChain.py b/src/backend/langflow/components/chains/ConversationChain.py index 7d9d28dcc..726056138 100644 --- a/src/backend/langflow/components/chains/ConversationChain.py +++ b/src/backend/langflow/components/chains/ConversationChain.py @@ -31,7 +31,7 @@ class ConversationChainComponent(CustomComponent): chain = ConversationChain(llm=llm) else: chain = ConversationChain(llm=llm, memory=memory) - result = chain.invoke(inputs) + result = chain.invoke(input_value) # result is an AIMessage which is a subclass of BaseMessage # We need to check if it is a string or a BaseMessage if hasattr(result, "content") and isinstance(result.content, str): diff --git a/src/backend/langflow/components/io/ChatOutput.py b/src/backend/langflow/components/io/ChatOutput.py index e4c5d03ee..842a083fb 100644 --- a/src/backend/langflow/components/io/ChatOutput.py +++ b/src/backend/langflow/components/io/ChatOutput.py @@ -1,11 +1,11 @@ from typing import Optional, Union -from langflow import CustomComponent from langflow.field_typing import Text +from langflow.io.schema import ChatComponent from langflow.schema import Record -class ChatOutput(CustomComponent): +class ChatOutput(ChatComponent): display_name = "Chat Output" description = "Used to send a message to the chat." diff --git a/src/backend/langflow/components/models/AmazonBedrockModel.py b/src/backend/langflow/components/models/AmazonBedrockModel.py index 68e404773..478bf8e9a 100644 --- a/src/backend/langflow/components/models/AmazonBedrockModel.py +++ b/src/backend/langflow/components/models/AmazonBedrockModel.py @@ -60,7 +60,7 @@ class AmazonBedrockComponent(CustomComponent): ) # type: ignore except Exception as e: raise ValueError("Could not connect to AmazonBedrock API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/AnthropicModel.py b/src/backend/langflow/components/models/AnthropicModel.py index be6e46d9a..cb8e55194 100644 --- a/src/backend/langflow/components/models/AnthropicModel.py +++ b/src/backend/langflow/components/models/AnthropicModel.py @@ -77,7 +77,7 @@ class AnthropicLLM(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to Anthropic API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/AzureOpenAIModel.py b/src/backend/langflow/components/models/AzureOpenAIModel.py index be1f724bf..df8413870 100644 --- a/src/backend/langflow/components/models/AzureOpenAIModel.py +++ b/src/backend/langflow/components/models/AzureOpenAIModel.py @@ -99,7 +99,7 @@ class AzureChatOpenAIComponent(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to AzureOpenAI API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/BaiduQianfanChatModel.py b/src/backend/langflow/components/models/BaiduQianfanChatModel.py index 9eadb7013..0075316a2 100644 --- a/src/backend/langflow/components/models/BaiduQianfanChatModel.py +++ b/src/backend/langflow/components/models/BaiduQianfanChatModel.py @@ -94,7 +94,7 @@ class QianfanChatEndpointComponent(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to Baidu Qianfan API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/CTransformersModel.py b/src/backend/langflow/components/models/CTransformersModel.py index 60cc2eb12..fb292d9cf 100644 --- a/src/backend/langflow/components/models/CTransformersModel.py +++ b/src/backend/langflow/components/models/CTransformersModel.py @@ -42,7 +42,7 @@ class CTransformersComponent(CustomComponent): output = CTransformers( model=model, model_file=model_file, model_type=model_type, config=config ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/CohereModel.py b/src/backend/langflow/components/models/CohereModel.py index 28b198ec1..c2a004c38 100644 --- a/src/backend/langflow/components/models/CohereModel.py +++ b/src/backend/langflow/components/models/CohereModel.py @@ -43,7 +43,7 @@ class CohereComponent(CustomComponent): max_tokens=max_tokens, temperature=temperature, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/GoogleGenerativeAIModel.py b/src/backend/langflow/components/models/GoogleGenerativeAIModel.py index 2ff01c4c7..3b0c758e8 100644 --- a/src/backend/langflow/components/models/GoogleGenerativeAIModel.py +++ b/src/backend/langflow/components/models/GoogleGenerativeAIModel.py @@ -73,7 +73,7 @@ class GoogleGenerativeAIComponent(CustomComponent): n=n or 1, google_api_key=SecretStr(google_api_key), ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/HuggingFaceModel.py b/src/backend/langflow/components/models/HuggingFaceModel.py index 394938344..99fcc6ab0 100644 --- a/src/backend/langflow/components/models/HuggingFaceModel.py +++ b/src/backend/langflow/components/models/HuggingFaceModel.py @@ -45,7 +45,7 @@ class HuggingFaceEndpointsComponent(CustomComponent): except Exception as e: raise ValueError("Could not connect to HuggingFace Endpoints API.") from e output = ChatHuggingFace(llm=llm) - message = output.invoke(inputs) + message = output.invoke(input_value)alue) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/LlamaCppModel.py b/src/backend/langflow/components/models/LlamaCppModel.py index 53a6f8ace..5941f213b 100644 --- a/src/backend/langflow/components/models/LlamaCppModel.py +++ b/src/backend/langflow/components/models/LlamaCppModel.py @@ -135,7 +135,7 @@ class LlamaCppComponent(CustomComponent): verbose=verbose, vocab_only=vocab_only, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/OllamaModel.py b/src/backend/langflow/components/models/OllamaModel.py index 3dc8dacab..7ae896532 100644 --- a/src/backend/langflow/components/models/OllamaModel.py +++ b/src/backend/langflow/components/models/OllamaModel.py @@ -250,7 +250,7 @@ class ChatOllamaComponent(CustomComponent): output = ChatOllama(**llm_params) # type: ignore except Exception as e: raise ValueError("Could not initialize Ollama LLM.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/OpenAIModel.py b/src/backend/langflow/components/models/OpenAIModel.py index 07ba7013c..ee6809e0b 100644 --- a/src/backend/langflow/components/models/OpenAIModel.py +++ b/src/backend/langflow/components/models/OpenAIModel.py @@ -80,7 +80,7 @@ class OpenAIModelComponent(CustomComponent): temperature=temperature, ) - message = model.invoke(inputs) + message = model.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/VertexAiModel.py b/src/backend/langflow/components/models/VertexAiModel.py index 81338f723..c05dc5e94 100644 --- a/src/backend/langflow/components/models/VertexAiModel.py +++ b/src/backend/langflow/components/models/VertexAiModel.py @@ -92,7 +92,7 @@ class ChatVertexAIComponent(CustomComponent): top_p=top_p, verbose=verbose, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/graph/edge/base.py b/src/backend/langflow/graph/edge/base.py index 99b24e2da..cfcd33dd1 100644 --- a/src/backend/langflow/graph/edge/base.py +++ b/src/backend/langflow/graph/edge/base.py @@ -4,6 +4,7 @@ from loguru import logger from pydantic import BaseModel, Field from langflow.graph.edge.utils import build_clean_params +from langflow.graph.schema import INPUT_FIELD_NAME from langflow.services.deps import get_monitor_service from langflow.services.monitor.utils import log_message @@ -12,7 +13,9 @@ if TYPE_CHECKING: class SourceHandle(BaseModel): - baseClasses: List[str] = Field(..., description="List of base classes for the source handle.") + baseClasses: List[str] = Field( + ..., description="List of base classes for the source handle." + ) dataType: str = Field(..., description="Data type for the source handle.") id: str = Field(..., description="Unique identifier for the source handle.") @@ -20,7 +23,9 @@ class SourceHandle(BaseModel): class TargetHandle(BaseModel): fieldName: str = Field(..., description="Field name for the target handle.") id: str = Field(..., description="Unique identifier for the target handle.") - inputTypes: Optional[List[str]] = Field(None, description="List of input types for the target handle.") + inputTypes: Optional[List[str]] = Field( + None, description="List of input types for the target handle." + ) type: str = Field(..., description="Type of the target handle.") @@ -49,16 +54,24 @@ class Edge: def validate_handles(self, source, target) -> None: if self.target_handle.inputTypes is None: - self.valid_handles = self.target_handle.type in self.source_handle.baseClasses + self.valid_handles = ( + self.target_handle.type in self.source_handle.baseClasses + ) else: self.valid_handles = ( - any(baseClass in self.target_handle.inputTypes for baseClass in self.source_handle.baseClasses) + any( + baseClass in self.target_handle.inputTypes + for baseClass in self.source_handle.baseClasses + ) or self.target_handle.type in self.source_handle.baseClasses ) if not self.valid_handles: logger.debug(self.source_handle) logger.debug(self.target_handle) - raise ValueError(f"Edge between {source.vertex_type} and {target.vertex_type} " f"has invalid handles") + raise ValueError( + f"Edge between {source.vertex_type} and {target.vertex_type} " + f"has invalid handles" + ) def __setstate__(self, state): self.source_id = state["source_id"] @@ -75,7 +88,11 @@ class Edge: # Both lists contain strings and sometimes a string contains the value we are # looking for e.g. comgin_out=["Chain"] and target_reqs=["LLMChain"] # so we need to check if any of the strings in source_types is in target_reqs - self.valid = any(output in target_req for output in self.source_types for target_req in self.target_reqs) + self.valid = any( + output in target_req + for output in self.source_types + for target_req in self.target_reqs + ) # Get what type of input the target node is expecting self.matched_type = next( @@ -86,7 +103,10 @@ class Edge: if no_matched_type: logger.debug(self.source_types) logger.debug(self.target_reqs) - raise ValueError(f"Edge between {source.vertex_type} and {target.vertex_type} " f"has no matched type") + raise ValueError( + f"Edge between {source.vertex_type} and {target.vertex_type} " + f"has no matched type" + ) def __repr__(self) -> str: return ( @@ -98,7 +118,11 @@ class Edge: return hash(self.__repr__()) def __eq__(self, __value: object) -> bool: - return self.__repr__() == __value.__repr__() if isinstance(__value, Edge) else False + return ( + self.__repr__() == __value.__repr__() + if isinstance(__value, Edge) + else False + ) class ContractEdge(Edge): @@ -137,14 +161,15 @@ class ContractEdge(Edge): log_transaction(self, source, target, "success") # If the target vertex is a power component we log messages if target.vertex_type == "ChatOutput" and ( - isinstance(target.params.get("message"), str) or isinstance(target.params.get("message"), dict) + isinstance(target.params.get(INPUT_FIELD_NAME), str) + or isinstance(target.params.get(INPUT_FIELD_NAME), dict) ): if target.params.get("message") == "": return self.result await log_message( sender=target.params.get("sender", ""), sender_name=target.params.get("sender_name", ""), - message=target.params.get("message", {}), + message=target.params.get(INPUT_FIELD_NAME, {}), session_id=target.params.get("session_id", ""), artifacts=target.artifacts, ) @@ -154,7 +179,9 @@ class ContractEdge(Edge): return f"{self.source_id} -[{self.target_param}]-> {self.target_id}" -def log_transaction(edge: ContractEdge, source: "Vertex", target: "Vertex", status, error=None): +def log_transaction( + edge: ContractEdge, source: "Vertex", target: "Vertex", status, error=None +): try: monitor_service = get_monitor_service() clean_params = build_clean_params(target) diff --git a/src/backend/langflow/graph/graph/base.py b/src/backend/langflow/graph/graph/base.py index 341a4729c..3f5e376a5 100644 --- a/src/backend/langflow/graph/graph/base.py +++ b/src/backend/langflow/graph/graph/base.py @@ -8,7 +8,7 @@ from loguru import logger from langflow.graph.edge.base import ContractEdge from langflow.graph.graph.constants import lazy_load_vertex_dict from langflow.graph.graph.utils import process_flow -from langflow.graph.schema import InterfaceComponentTypes +from langflow.graph.schema import INPUT_FIELD_NAME, InterfaceComponentTypes from langflow.graph.vertex.base import Vertex from langflow.graph.vertex.types import ( ChatVertex, @@ -33,8 +33,6 @@ class Graph: edges: List[Dict[str, str]], flow_id: Optional[str] = None, ) -> None: - self.inputs = [] - self.outputs = [] self._vertices = nodes self._edges = edges self.raw_graph_data = {"nodes": nodes, "edges": edges} @@ -77,7 +75,7 @@ class Graph: async def _run(self, inputs: Dict[str, str]) -> List["ResultData"]: """Runs the graph with the given inputs.""" - for vertex_id in self.inputs: + for vertex_id in self._is_input_vertices: vertex = self.get_vertex(vertex_id) if vertex is None: raise ValueError(f"Vertex {vertex_id} not found") @@ -89,7 +87,7 @@ class Graph: logger.exception(exc) raise ValueError(f"Error running graph: {exc}") from exc outputs = [] - for vertex_id in self.outputs: + for vertex_id in self._is_output_vertices: vertex = self.get_vertex(vertex_id) if vertex is None: raise ValueError(f"Vertex {vertex_id} not found") @@ -104,11 +102,11 @@ class Graph: # of the vertices that are inputs # if the value is a list, we need to run multiple times outputs = [] - inputs_values = inputs.get("input_value") + inputs_values = inputs.get(INPUT_FIELD_NAME) if not isinstance(inputs_values, list): inputs_values = [inputs_values] for input_value in inputs_values: - run_outputs = await self._run({"input_value": input_value}) + run_outputs = await self._run({INPUT_FIELD_NAME: input_value}) logger.debug(f"Run outputs: {run_outputs}") outputs.extend(run_outputs) return outputs @@ -317,28 +315,6 @@ class Graph: # Now that we have the vertices and edges # We need to map the vertices that are connected to # to ChatVertex instances - self._map_chat_vertices() - - def _map_chat_vertices(self) -> None: - """Maps the vertices that are connected to ChatVertex instances.""" - # For each edge, we need to check if the source or target vertex is a ChatVertex - # If it is, we need to update the other vertex `is_external` attribute - # and store the id of the ChatVertex in the attributes self.inputs and self.outputs - for edge in self.edges: - source_vertex = self.get_vertex(edge.source_id) - target_vertex = self.get_vertex(edge.target_id) - if isinstance(source_vertex, ChatVertex): - # The source vertex is a ChatVertex - # thus the target vertex is an external vertex - # and the source vertex is an input - target_vertex.has_external_input = True - self.inputs.append(source_vertex.id) - if isinstance(target_vertex, ChatVertex): - # The target vertex is a ChatVertex - # thus the source vertex is an external vertex - # and the target vertex is an output - source_vertex.has_external_output = True - self.outputs.append(target_vertex.id) def remove_vertex(self, vertex_id: str) -> None: """Removes a vertex from the graph.""" @@ -443,13 +419,15 @@ class Graph: async def _execute_tasks(self, tasks): """Executes tasks in parallel, handling exceptions for each task.""" results = [] - for task in asyncio.as_completed(tasks): + for i, task in enumerate(asyncio.as_completed(tasks)): try: result = await task results.append(result) except Exception as e: # Log the exception along with the task name for easier debugging - task_name = task.get_name() + # task_name = task.get_name() + # coroutine has not attribute get_name + task_name = tasks[i].get_name() logger.error(f"Task {task_name} failed with exception: {e}") return results diff --git a/src/backend/langflow/graph/schema.py b/src/backend/langflow/graph/schema.py index d41e0544a..028b8db9f 100644 --- a/src/backend/langflow/graph/schema.py +++ b/src/backend/langflow/graph/schema.py @@ -35,3 +35,5 @@ OUTPUT_COMPONENTS = [ InterfaceComponentTypes.ChatOutput, InterfaceComponentTypes.TextOutput, ] + +INPUT_FIELD_NAME = "input_value" diff --git a/src/backend/langflow/graph/vertex/base.py b/src/backend/langflow/graph/vertex/base.py index 3e1133491..dd308f9f1 100644 --- a/src/backend/langflow/graph/vertex/base.py +++ b/src/backend/langflow/graph/vertex/base.py @@ -44,7 +44,7 @@ class Vertex: ) -> None: # is_external means that the Vertex send or receives data from # an external source (e.g the chat) - + self.updated_raw_params = False self.id: str = data["id"] self.is_input = any( input_component_name in self.id for input_component_name in INPUT_COMPONENTS @@ -285,6 +285,10 @@ class Vertex: if self.graph is None: raise ValueError("Graph not found") + if self.updated_raw_params: + self.updated_raw_params = False + return + template_dict = { key: value for key, value in self.data["node"]["template"].items() @@ -386,10 +390,11 @@ class Vertex: Raises: ValueError: If any key in new_params is not found in self._raw_params. """ - for key in new_params: - if key not in self._raw_params: - raise ValueError(f"Key {key} not found in raw params") + # First check if the input_value in _raw_params is not a vertex + if any(isinstance(self._raw_params.get(key), Vertex) for key in new_params): + return self._raw_params.update(new_params) + self.updated_raw_params = True async def _build(self, user_id=None): """ @@ -451,6 +456,8 @@ class Vertex: await self._build_node_and_update_params(key, value, user_id) elif isinstance(value, list) and self._is_list_of_nodes(value): await self._build_list_of_nodes_and_update_params(key, value, user_id) + elif key not in self.params: + self.params[key] = value def _is_node(self, value): """ @@ -586,7 +593,7 @@ class Vertex: logger.warning(message) - def _reset(self): + def _reset(self, params_update: Optional[Dict[str, Any]] = None): self._built = False self._built_object = UnbuiltObject() self._built_result = UnbuiltResult() diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index 45bcb9ccd..100390d11 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -1,11 +1,12 @@ import ast import json -from typing import (AsyncIterator, Callable, Dict, Iterator, List, Optional, - Union) +from typing import AsyncIterator, Callable, Dict, Iterator, List, Optional, Union import yaml from langchain_core.messages import AIMessage +from loguru import logger +from langflow.graph.schema import INPUT_FIELD_NAME from langflow.graph.utils import UnbuiltObject, flatten_list from langflow.graph.vertex.base import StatefulVertex, StatelessVertex from langflow.interface.utils import extract_input_variables_from_prompt @@ -362,7 +363,7 @@ class ChatVertex(StatelessVertex): artifacts = None sender = self.params.get("sender", None) sender_name = self.params.get("sender_name", None) - message = self.params.get("message", None) + message = self.params.get(INPUT_FIELD_NAME, None) stream_url = None if isinstance(self._built_object, AIMessage): artifacts = ChatOutputResponse.from_message( @@ -396,7 +397,7 @@ class ChatVertex(StatelessVertex): if artifacts: self.artifacts = artifacts.model_dump() if isinstance(self._built_object, (AsyncIterator, Iterator)): - if self.params["as_record"]: + if self.params["return_record"]: self._built_object = Record(text=message, data=self.artifacts) else: self._built_object = message @@ -406,7 +407,7 @@ class ChatVertex(StatelessVertex): await super()._run(*args, **kwargs) async def stream(self): - iterator = self.params.get("message", None) + iterator = self.params.get(INPUT_FIELD_NAME, None) if not isinstance(iterator, (AsyncIterator, Iterator)): raise ValueError("The message must be an iterator or an async iterator.") is_async = isinstance(iterator, AsyncIterator) @@ -423,15 +424,17 @@ class ChatVertex(StatelessVertex): message = message.text if hasattr(message, "text") else message yield message complete_message += message - self._built_object = Record(text=complete_message, data=self.artifacts) - self._built_result = complete_message - # Update artifacts with the message - # and remove the stream_url self.artifacts = ChatOutputResponse( message=complete_message, sender=self.params.get("sender", ""), sender_name=self.params.get("sender_name", ""), ).model_dump() + self.params[INPUT_FIELD_NAME] = complete_message + self._built_object = Record(text=complete_message, data=self.artifacts) + self._built_result = complete_message + # Update artifacts with the message + # and remove the stream_url + logger.debug(f"Streamed message: {complete_message}") await log_message( sender=self.params.get("sender", ""), @@ -441,6 +444,9 @@ class ChatVertex(StatelessVertex): artifacts=self.artifacts, ) + self._validate_built_object() + self._built = True + class RoutingVertex(StatelessVertex): def __init__(self, data: Dict, graph): diff --git a/src/backend/langflow/interface/custom/code_parser/code_parser.py b/src/backend/langflow/interface/custom/code_parser/code_parser.py index 7a102e33f..e54051a5c 100644 --- a/src/backend/langflow/interface/custom/code_parser/code_parser.py +++ b/src/backend/langflow/interface/custom/code_parser/code_parser.py @@ -6,6 +6,9 @@ from typing import Any, Dict, List, Type, Union from cachetools import TTLCache, cachedmethod, keys from fastapi import HTTPException +from loguru import logger + +from langflow.interface.custom.eval import eval_custom_component_code from langflow.interface.custom.schema import CallableCodeDetails, ClassCodeDetails @@ -92,7 +95,9 @@ class CodeParser: elif isinstance(node, ast.ImportFrom): for alias in node.names: if alias.asname: - self.data["imports"].append((node.module, f"{alias.name} as {alias.asname}")) + self.data["imports"].append( + (node.module, f"{alias.name} as {alias.asname}") + ) else: self.data["imports"].append((node.module, alias.name)) @@ -141,7 +146,9 @@ class CodeParser: return_type = None if node.returns: return_type_str = ast.unparse(node.returns) - eval_env = self.construct_eval_env(return_type_str, tuple(self.data["imports"])) + eval_env = self.construct_eval_env( + return_type_str, tuple(self.data["imports"]) + ) try: return_type = eval(return_type_str, eval_env) @@ -183,14 +190,22 @@ class CodeParser: num_defaults = len(node.args.defaults) num_missing_defaults = num_args - num_defaults missing_defaults = [None] * num_missing_defaults - default_values = [ast.unparse(default).strip("'") if default else None for default in node.args.defaults] + default_values = [ + ast.unparse(default).strip("'") if default else None + for default in node.args.defaults + ] # Now check all default values to see if there # are any "None" values in the middle - default_values = [None if value == "None" else value for value in default_values] + default_values = [ + None if value == "None" else value for value in default_values + ] defaults = missing_defaults + default_values - args = [self.parse_arg(arg, default) for arg, default in zip(node.args.args, defaults)] + args = [ + self.parse_arg(arg, default) + for arg, default in zip(node.args.args, defaults) + ] return args def parse_varargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]: @@ -208,11 +223,17 @@ class CodeParser: """ Parses the keyword-only arguments of a function or method node. """ - kw_defaults = [None] * (len(node.args.kwonlyargs) - len(node.args.kw_defaults)) + [ - ast.unparse(default) if default else None for default in node.args.kw_defaults + kw_defaults = [None] * ( + len(node.args.kwonlyargs) - len(node.args.kw_defaults) + ) + [ + ast.unparse(default) if default else None + for default in node.args.kw_defaults ] - args = [self.parse_arg(arg, default) for arg, default in zip(node.args.kwonlyargs, kw_defaults)] + args = [ + self.parse_arg(arg, default) + for arg, default in zip(node.args.kwonlyargs, kw_defaults) + ] return args def parse_kwargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]: @@ -268,15 +289,28 @@ class CodeParser: method = self.parse_callable_details(stmt) return (method, True) if stmt.name == "__init__" else (method, False) + def get_base_classes(self): + """ + Returns the base classes of the custom component class. + """ + try: + bases = self.execute_and_inspect_classes(self.code) + except Exception as e: + # If the code cannot be executed, return an empty list + logger.exception(e) + bases = [] + raise e + return bases + def parse_classes(self, node: ast.ClassDef) -> None: """ Extracts "classes" from the code, including inheritance and init methods. """ - + bases = self.get_base_classes() or [ast.unparse(b) for b in node.bases] class_details = ClassCodeDetails( name=node.name, doc=ast.get_docstring(node), - bases=[ast.unparse(base) for base in node.bases], + bases=bases, attributes=[], methods=[], init=None, @@ -303,11 +337,25 @@ class CodeParser: Extracts global variables from the code. """ global_var = { - "targets": [t.id if hasattr(t, "id") else ast.dump(t) for t in node.targets], + "targets": [ + t.id if hasattr(t, "id") else ast.dump(t) for t in node.targets + ], "value": ast.unparse(node.value), } self.data["global_vars"].append(global_var) + def execute_and_inspect_classes(self, code: str): + custom_component_class = eval_custom_component_code(code) + custom_component = custom_component_class() + dunder_class = custom_component.__class__ + # Get the base classes at two levels of inheritance + bases = [] + for base in dunder_class.__bases__: + bases.append(base.__name__) + for bases_base in base.__bases__: + bases.append(bases_base.__name__) + return bases + def parse_code(self) -> Dict[str, Any]: """ Runs all parsing operations and returns the resulting data. diff --git a/src/backend/langflow/interface/custom/utils.py b/src/backend/langflow/interface/custom/utils.py index d2be47333..381193116 100644 --- a/src/backend/langflow/interface/custom/utils.py +++ b/src/backend/langflow/interface/custom/utils.py @@ -27,14 +27,18 @@ from langflow.utils import validate from langflow.utils.util import get_base_classes -def add_output_types(frontend_node: CustomComponentFrontendNode, return_types: List[str]): +def add_output_types( + frontend_node: CustomComponentFrontendNode, return_types: List[str] +): """Add output types to the frontend node""" for return_type in return_types: if return_type is None: raise HTTPException( status_code=400, detail={ - "error": ("Invalid return type. Please check your code and try again."), + "error": ( + "Invalid return type. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) @@ -63,14 +67,18 @@ def reorder_fields(frontend_node: CustomComponentFrontendNode, field_order: List frontend_node.template.fields = reordered_fields -def add_base_classes(frontend_node: CustomComponentFrontendNode, return_types: List[str]): +def add_base_classes( + frontend_node: CustomComponentFrontendNode, return_types: List[str] +): """Add base classes to the frontend node""" for return_type_instance in return_types: if return_type_instance is None: raise HTTPException( status_code=400, detail={ - "error": ("Invalid return type. Please check your code and try again."), + "error": ( + "Invalid return type. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) @@ -145,10 +153,14 @@ def add_new_custom_field( # If options is a list, then it's a dropdown # If options is None, then it's a list of strings is_list = isinstance(field_config.get("options"), list) - field_config["is_list"] = is_list or field_config.get("is_list", False) or field_contains_list + field_config["is_list"] = ( + is_list or field_config.get("is_list", False) or field_contains_list + ) if "name" in field_config: - warnings.warn("The 'name' key in field_config is used to build the object and can't be changed.") + warnings.warn( + "The 'name' key in field_config is used to build the object and can't be changed." + ) required = field_config.pop("required", field_required) placeholder = field_config.pop("placeholder", "") @@ -179,7 +191,9 @@ def add_extra_fields(frontend_node, field_config, function_args): if "name" not in extra_field or extra_field["name"] == "self": continue - field_name, field_type, field_value, field_required = get_field_properties(extra_field) + field_name, field_type, field_value, field_required = get_field_properties( + extra_field + ) config = field_config.get(field_name, {}) frontend_node = add_new_custom_field( frontend_node, @@ -217,7 +231,9 @@ def run_build_config( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -245,7 +261,9 @@ def run_build_config( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -300,16 +318,24 @@ def build_custom_component_template( frontend_node = build_frontend_node(custom_component.template_config) logger.debug("Updated attributes") - field_config, custom_instance = run_build_config(custom_component, user_id=user_id, update_field=update_field) + field_config, custom_instance = run_build_config( + custom_component, user_id=user_id, update_field=update_field + ) logger.debug("Built field config") entrypoint_args = custom_component.get_function_entrypoint_args add_extra_fields(frontend_node, field_config, entrypoint_args) - frontend_node = add_code_field(frontend_node, custom_component.code, field_config.get("code", {})) + frontend_node = add_code_field( + frontend_node, custom_component.code, field_config.get("code", {}) + ) - add_base_classes(frontend_node, custom_component.get_function_entrypoint_return_type) - add_output_types(frontend_node, custom_component.get_function_entrypoint_return_type) + add_base_classes( + frontend_node, custom_component.get_function_entrypoint_return_type + ) + add_output_types( + frontend_node, custom_component.get_function_entrypoint_return_type + ) logger.debug("Added base classes") reorder_fields(frontend_node, custom_instance._get_field_order()) @@ -321,7 +347,9 @@ def build_custom_component_template( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -345,7 +373,9 @@ def build_custom_components(settings_service): if not settings_service.settings.COMPONENTS_PATH: return {} - logger.info(f"Building custom components from {settings_service.settings.COMPONENTS_PATH}") + logger.info( + f"Building custom components from {settings_service.settings.COMPONENTS_PATH}" + ) custom_components_from_file = {} processed_paths = set() for path in settings_service.settings.COMPONENTS_PATH: @@ -356,7 +386,9 @@ def build_custom_components(settings_service): custom_component_dict = build_custom_component_list_from_path(path_str) if custom_component_dict: category = next(iter(custom_component_dict)) - logger.info(f"Loading {len(custom_component_dict[category])} component(s) from category {category}") + logger.info( + f"Loading {len(custom_component_dict[category])} component(s) from category {category}" + ) custom_components_from_file = merge_nested_dicts_with_renaming( custom_components_from_file, custom_component_dict ) @@ -400,8 +432,9 @@ def sanitize_field_config(field_config: Dict): def build_component(component): """Build a single component.""" component_name = determine_component_name(component) + logger.debug(f"Building component: {component_name}") component_template = create_component_template(component) - logger.debug(f"Building component: {component_name, component.get('output_types')}") + return component_name, component_template diff --git a/src/backend/langflow/io/schema.py b/src/backend/langflow/io/schema.py index d83d3472e..4660b4276 100644 --- a/src/backend/langflow/io/schema.py +++ b/src/backend/langflow/io/schema.py @@ -48,27 +48,28 @@ class ChatComponent(CustomComponent): raise ValueError( "All of session_id, sender, and sender_name must be provided." ) - - if not record: - record = [] - if not session_id or not sender or not sender_name: - raise ValueError - for text in text: - record = Record( - text=text, - data={ - "session_id": session_id, - "sender": sender, - "sender_name": sender_name, - }, - ) - record.append(record) - elif isinstance(record, Record): - record = [record] + if isinstance(message, Record): + record = message + record.data.update( + { + "session_id": session_id, + "sender": sender, + "sender_name": sender_name, + } + ) + else: + record = Record( + text=message, + data={ + "session_id": session_id, + "sender": sender, + "sender_name": sender_name, + }, + ) self.status = record - record = add_messages(record) - return record + records = add_messages([record]) + return records[0] def build( self, diff --git a/src/backend/langflow/processing/process.py b/src/backend/langflow/processing/process.py index 69e47b242..aa419b962 100644 --- a/src/backend/langflow/processing/process.py +++ b/src/backend/langflow/processing/process.py @@ -269,6 +269,7 @@ async def build_graph_and_generate_result( async def run_graph( graph: Union["Graph", dict], + flow_id: str, session_id: str, inputs: Optional[Union[dict, List[dict]]] = None, artifacts: Optional[Dict[str, Any]] = None, @@ -276,7 +277,8 @@ async def run_graph( ): """Run the graph and generate the result""" if isinstance(graph, dict): - graph = Graph.from_payload(graph) + graph = Graph.from_payload(graph, flow_id=flow_id) + outputs = await graph.run(inputs) if session_id and session_service: session_service.update_session(session_id, (graph, artifacts)) diff --git a/src/backend/langflow/utils/logger.py b/src/backend/langflow/utils/logger.py index 060ad9731..6755e53b1 100644 --- a/src/backend/langflow/utils/logger.py +++ b/src/backend/langflow/utils/logger.py @@ -25,7 +25,10 @@ def patching(record): def configure(log_level: Optional[str] = None, log_file: Optional[Path] = None): - if os.getenv("LANGFLOW_LOG_LEVEL") in VALID_LOG_LEVELS and log_level is None: + if ( + os.getenv("LANGFLOW_LOG_LEVEL", "").upper() in VALID_LOG_LEVELS + and log_level is None + ): log_level = os.getenv("LANGFLOW_LOG_LEVEL") if log_level is None: log_level = "INFO" diff --git a/src/backend/langflow/utils/validate.py b/src/backend/langflow/utils/validate.py index 21821538c..6b789c08c 100644 --- a/src/backend/langflow/utils/validate.py +++ b/src/backend/langflow/utils/validate.py @@ -45,7 +45,9 @@ def validate_code(code): # Evaluate the function definition for node in tree.body: if isinstance(node, ast.FunctionDef): - code_obj = compile(ast.Module(body=[node], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[node], type_ignores=[]), "", "exec" + ) try: exec(code_obj) except Exception as e: @@ -89,15 +91,23 @@ def execute_function(code, function_name, *args, **kwargs): exec_globals, locals(), ) - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e function_code = next( - node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == function_name + node + for node in module.body + if isinstance(node, ast.FunctionDef) and node.name == function_name ) function_code.parent = None - code_obj = compile(ast.Module(body=[function_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[function_code], type_ignores=[]), "", "exec" + ) try: exec(code_obj, exec_globals, locals()) except Exception as exc: @@ -124,15 +134,23 @@ def create_function(code, function_name): if isinstance(node, ast.Import): for alias in node.names: try: - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e function_code = next( - node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == function_name + node + for node in module.body + if isinstance(node, ast.FunctionDef) and node.name == function_name ) function_code.parent = None - code_obj = compile(ast.Module(body=[function_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[function_code], type_ignores=[]), "", "exec" + ) with contextlib.suppress(Exception): exec(code_obj, exec_globals, locals()) exec_globals[function_name] = locals()[function_name] @@ -194,9 +212,13 @@ def prepare_global_scope(code, module): if isinstance(node, ast.Import): for alias in node.names: try: - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e elif isinstance(node, ast.ImportFrom) and node.module is not None: try: imported_module = importlib.import_module(node.module) @@ -217,7 +239,11 @@ def extract_class_code(module, class_name): :param class_name: Name of the class to extract :return: AST node of the specified class """ - class_code = next(node for node in module.body if isinstance(node, ast.ClassDef) and node.name == class_name) + class_code = next( + node + for node in module.body + if isinstance(node, ast.ClassDef) and node.name == class_name + ) class_code.parent = None return class_code @@ -230,7 +256,9 @@ def compile_class_code(class_code): :param class_code: AST node of the class :return: Compiled code object of the class """ - code_obj = compile(ast.Module(body=[class_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[class_code], type_ignores=[]), "", "exec" + ) return code_obj @@ -274,7 +302,9 @@ def get_default_imports(code_string): langflow_imports = list(CUSTOM_COMPONENT_SUPPORTED_TYPES.keys()) necessary_imports = find_names_in_code(code_string, langflow_imports) langflow_module = importlib.import_module("langflow.field_typing") - default_imports.update({name: getattr(langflow_module, name) for name in necessary_imports}) + default_imports.update( + {name: getattr(langflow_module, name) for name in necessary_imports} + ) return default_imports diff --git a/src/frontend/src/components/newChatView/chatMessage/index.tsx b/src/frontend/src/components/newChatView/chatMessage/index.tsx index 266444d8b..4f6d8e534 100644 --- a/src/frontend/src/components/newChatView/chatMessage/index.tsx +++ b/src/frontend/src/components/newChatView/chatMessage/index.tsx @@ -1,5 +1,5 @@ import Convert from "ansi-to-html"; -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useState, useRef } from "react"; import Markdown from "react-markdown"; import rehypeMathjax from "rehype-mathjax"; import remarkGfm from "remark-gfm"; @@ -12,12 +12,14 @@ import IconComponent from "../../../components/genericIconComponent"; import { chatMessagePropsType } from "../../../types/components"; import { classNames } from "../../../utils/utils"; import FileCard from "../fileComponent"; +import useFlowStore from "../../../stores/flowStore"; export default function ChatMessage({ chat, lockChat, lastMessage, updateChat, + setLockChat }: chatMessagePropsType): JSX.Element { const convert = new Convert({ newline: true }); const [hidden, setHidden] = useState(true); @@ -29,6 +31,16 @@ export default function ChatMessage({ const chatMessageString = chat.message ? chat.message.toString() : ""; const [chatMessage, setChatMessage] = useState(chatMessageString); const [isStreaming, setIsStreaming] = useState(false); + const eventSource = useRef(undefined); + const updateFlowPool = useFlowStore((state) => state.updateFlowPool); + const chatMessageRef = useRef(chatMessage); + + // Sync ref with state + useEffect(() => { + chatMessageRef.current = chatMessage; + }, [chatMessage]); + + // The idea now is that chat.stream_url MAY be a URL if we should stream the output of the chat // probably the message is empty when we have a stream_url @@ -36,49 +48,52 @@ export default function ChatMessage({ const streamChunks = (url: string) => { setIsStreaming(true); // Streaming starts return new Promise((resolve, reject) => { - const eventSource = new EventSource(url); - eventSource.onmessage = (event) => { + eventSource.current = new EventSource(url); + eventSource.current.onmessage = (event) => { let parsedData = JSON.parse(event.data); if (parsedData.chunk) { setChatMessage((prev) => prev + parsedData.chunk); } }; - eventSource.onerror = (event) => { + eventSource.current.onerror = (event) => { + setIsStreaming(false); + eventSource.current?.close(); + setStreamUrl(undefined); reject(new Error("Streaming failed")); - setIsStreaming(false); - eventSource.close(); }; - eventSource.addEventListener("close", (event) => { - setStreamUrl(null); // Update state to reflect the stream is closed - resolve(true); + eventSource.current.addEventListener("close", (event) => { + setStreamUrl(undefined); // Update state to reflect the stream is closed + eventSource.current?.close(); setIsStreaming(false); - eventSource.close(); + resolve(true); }); }); }; + useEffect(() => { - if (streamUrl && chat.message === "") { + console.log("chatMessage", chatMessage); + if (streamUrl && !isStreaming) { + setLockChat(true); streamChunks(streamUrl) .then(() => { + setLockChat(false); if (updateChat) { - updateChat(chat, chatMessage, streamUrl); + updateChat(chat, chatMessageRef.current); } }) .catch((error) => { console.error(error); + setLockChat(false); }); } - }, [streamUrl]); + }, [streamUrl, chatMessage]); useEffect(() => { - // This effect is specifically for calling updateChat after streaming ends - if (!isStreaming && streamUrl) { - if (updateChat) { - updateChat(chat, chatMessage, streamUrl); - } + return () => { + eventSource.current?.close(); } - }, [isStreaming]); + }, []) useEffect(() => { const element = document.getElementById("last-chat-message"); @@ -207,7 +222,7 @@ dark:prose-invert" }, ]} activeTab={"0"} - setActiveTab={() => {}} + setActiveTab={() => { }} /> ) : ( @@ -264,33 +279,33 @@ dark:prose-invert" {promptOpen ? template?.split("\n")?.map((line, index) => { - const regex = /{([^}]+)}/g; - let match; - let parts: Array = []; - let lastIndex = 0; - while ((match = regex.exec(line)) !== null) { - // Push text up to the match - if (match.index !== lastIndex) { - parts.push(line.substring(lastIndex, match.index)); - } - // Push div with matched text - if (chat.message[match[1]]) { - parts.push( - - {chat.message[match[1]]} - - ); - } + const regex = /{([^}]+)}/g; + let match; + let parts: Array = []; + let lastIndex = 0; + while ((match = regex.exec(line)) !== null) { + // Push text up to the match + if (match.index !== lastIndex) { + parts.push(line.substring(lastIndex, match.index)); + } + // Push div with matched text + if (chat.message[match[1]]) { + parts.push( + + {chat.message[match[1]]} + + ); + } - // Update last index - lastIndex = regex.lastIndex; - } - // Push text after the last match - if (lastIndex !== line.length) { - parts.push(line.substring(lastIndex)); - } - return

{parts}

; - }) + // Update last index + lastIndex = regex.lastIndex; + } + // Push text after the last match + if (lastIndex !== line.length) { + parts.push(line.substring(lastIndex)); + } + return

{parts}

; + }) : chatMessage}
diff --git a/src/frontend/src/components/newChatView/index.tsx b/src/frontend/src/components/newChatView/index.tsx index 86b8ace6a..b1f888f1c 100644 --- a/src/frontend/src/components/newChatView/index.tsx +++ b/src/frontend/src/components/newChatView/index.tsx @@ -34,6 +34,7 @@ export default function NewChatView({ const inputIds = inputs.map((obj) => obj.id); const outputIds = outputs.map((obj) => obj.id); const outputTypes = outputs.map((obj) => obj.type); + const updateFlowPool = useFlowStore((state)=>state.updateFlowPool) useEffect(() => { if (!outputTypes.includes("ChatOutput")) { @@ -67,14 +68,12 @@ export default function NewChatView({ const { sender, message, sender_name, stream_url } = output.data .artifacts as ChatOutputType; - const componentId = output.id + index; - const is_ai = sender === "Machine" || sender === null; return { isSend: !is_ai, message: message, sender_name, - id: componentId, + componentId: output.id, stream_url: stream_url, }; } catch (e) { @@ -83,7 +82,7 @@ export default function NewChatView({ isSend: false, message: "Error parsing message", sender_name: "Error", - id: output.id + index, + componentId: output.id, }; } }); @@ -120,27 +119,24 @@ export default function NewChatView({ function updateChat( chat: ChatMessageType, message: string, - stream_url: string | null - ) { - if (message === "") return; - console.log(`updateChat: ${message}`); - console.log("chatHistory:", chatHistory); - chat.message = message; - chat.stream_url = stream_url; + stream_url?: string + ) { + if (message === "") return; + chat.message = message; // chat is one of the chatHistory - setChatHistory((oldChatHistory) => { - const index = oldChatHistory.findIndex((ch) => ch.id === chat.id); - - if (index === -1) return oldChatHistory; - let newChatHistory = _.cloneDeep(oldChatHistory); - newChatHistory = [ - ...newChatHistory.slice(0, index), - chat, - ...newChatHistory.slice(index + 1), - ]; - console.log("newChatHistory:", newChatHistory); - return newChatHistory; - }); + updateFlowPool(chat.componentId,{message,sender_name:chat.sender_name??"Bot",sender:"Machine"}) + // setChatHistory((oldChatHistory) => { + // const index = oldChatHistory.findIndex((ch) => ch.id === chat.id); + // if (index === -1) return oldChatHistory; + // let newChatHistory = _.cloneDeep(oldChatHistory); + // newChatHistory = [ + // ...newChatHistory.slice(0, index), + // chat, + // ...newChatHistory.slice(index + 1), + // ]; + // console.log("newChatHistory:", newChatHistory); + // return newChatHistory; + // }); } return ( @@ -164,10 +160,11 @@ export default function NewChatView({ {chatHistory?.length > 0 ? ( chatHistory.map((chat, index) => ( )) diff --git a/src/frontend/src/stores/flowStore.ts b/src/frontend/src/stores/flowStore.ts index 300378406..f9a211e86 100644 --- a/src/frontend/src/stores/flowStore.ts +++ b/src/frontend/src/stores/flowStore.ts @@ -19,7 +19,7 @@ import { sourceHandleType, targetHandleType, } from "../types/flow"; -import { FlowStoreType } from "../types/zustand/flow"; +import { ChatOutputType, FlowPoolObjectType, FlowStoreType, chatInputType } from "../types/zustand/flow"; import { buildVertices } from "../utils/buildUtils"; import { cleanEdges, @@ -51,7 +51,7 @@ const useFlowStore = create((set, get) => ({ setFlowPool: (flowPool) => { set({ flowPool }); }, - addDataToFlowPool: (data: any, nodeId: string) => { + addDataToFlowPool: (data: FlowPoolObjectType, nodeId: string) => { let newFlowPool = cloneDeep({ ...get().flowPool }); if (!newFlowPool[nodeId]) newFlowPool[nodeId] = [data]; else { @@ -59,6 +59,28 @@ const useFlowStore = create((set, get) => ({ } get().setFlowPool(newFlowPool); }, + updateFlowPool:(nodeId:string,data:FlowPoolObjectType| ChatOutputType | chatInputType,buildId?:string)=>{ + let newFlowPool = cloneDeep({ ...get().flowPool }); + if (!newFlowPool[nodeId]){ + return; + } + else { + let index = newFlowPool[nodeId].length-1; + if(buildId){ + index = newFlowPool[nodeId].findIndex((flow)=>flow.id===buildId); + } + //check if the data is a flowpool object + if((data as FlowPoolObjectType).data?.artifacts!==undefined){ + newFlowPool[nodeId][index] = (data as FlowPoolObjectType); + } + //update data artifact + else + { + newFlowPool[nodeId][index].data.artifacts = data; + } + } + get().setFlowPool(newFlowPool); + }, CleanFlowPool: () => { get().setFlowPool({}); }, @@ -394,12 +416,13 @@ const useFlowStore = create((set, get) => ({ } function handleBuildUpdate( vertexBuildData: VertexBuildTypeAPI, - status: BuildStatus + status: BuildStatus, + buildId:string ) { if (vertexBuildData && vertexBuildData.inactive_vertices) { get().removeFromVerticesBuild(vertexBuildData.inactive_vertices); } - get().addDataToFlowPool(vertexBuildData, vertexBuildData.id); + get().addDataToFlowPool({...vertexBuildData,buildId}, vertexBuildData.id); useFlowStore.getState().updateBuildStatus([vertexBuildData.id], status); } await updateFlowInDatabase({ diff --git a/src/frontend/src/types/chat/index.ts b/src/frontend/src/types/chat/index.ts index 3914dd2dc..e24c6e891 100644 --- a/src/frontend/src/types/chat/index.ts +++ b/src/frontend/src/types/chat/index.ts @@ -9,7 +9,7 @@ export type ChatMessageType = { files?: Array<{ data: string; type: string; data_type: string }>; prompt?: string; chatKey?: string; - id?: string; + componentId: string; stream_url?: string | null; sender_name?: string; }; diff --git a/src/frontend/src/types/components/index.ts b/src/frontend/src/types/components/index.ts index e7767eabb..30a32ba19 100644 --- a/src/frontend/src/types/components/index.ts +++ b/src/frontend/src/types/components/index.ts @@ -524,10 +524,11 @@ export type chatMessagePropsType = { chat: ChatMessageType; lockChat: boolean; lastMessage: boolean; + setLockChat: (lock: boolean) => void; updateChat: ( chat: ChatMessageType, message: string, - stream_url: string + stream_url?: string ) => void; }; @@ -632,9 +633,9 @@ export type validationStatusType = { id: string; data: object | any; params: string; - progress: number; + progress?: number; valid: boolean; - duration: string; + duration?: string; }; export type ApiKey = { diff --git a/src/frontend/src/types/zustand/flow/index.ts b/src/frontend/src/types/zustand/flow/index.ts index cf6e786e0..c506b034f 100644 --- a/src/frontend/src/types/zustand/flow/index.ts +++ b/src/frontend/src/types/zustand/flow/index.ts @@ -18,16 +18,18 @@ export type ChatOutputType = { message: string; sender: string; sender_name: string; + stream_url?: string; }; export type FlowPoolObjectType = { timestamp: string; valid: boolean; params: any; - data: { artifacts: any; results: any | ChatOutputType | chatInputType }; - duration: string; - progress: number; + data: { artifacts: any | ChatOutputType | chatInputType; results: any | ChatOutputType | chatInputType }; + duration?: string; + progress?: number; id: string; + buildId: string; }; export type FlowPoolType = { @@ -40,7 +42,7 @@ export type FlowStoreType = { outputs: Array<{ type: string; id: string }>; hasIO: boolean; setFlowPool: (flowPool: FlowPoolType) => void; - addDataToFlowPool: (data: any, nodeId: string) => void; + addDataToFlowPool: (data: FlowPoolObjectType, nodeId: string) => void; CleanFlowPool: () => void; isBuilding: boolean; isPending: boolean; @@ -92,4 +94,5 @@ export type FlowStoreType = { updateBuildStatus: (nodeId: string[], status: BuildStatus) => void; revertBuiltStatusFromBuilding: () => void; flowBuildStatus: { [key: string]: BuildStatus }; + updateFlowPool: (nodeId:string, data:FlowPoolObjectType | ChatOutputType | chatInputType,buildId?:string) => void; }; diff --git a/src/frontend/src/utils/buildUtils.ts b/src/frontend/src/utils/buildUtils.ts index 862472ee7..3acc3d972 100644 --- a/src/frontend/src/utils/buildUtils.ts +++ b/src/frontend/src/utils/buildUtils.ts @@ -9,7 +9,7 @@ type BuildVerticesParams = { flowId: string; // Assuming FlowType is the type for your flow nodeId?: string | null; // Assuming nodeId is of type string, and it's optional onGetOrderSuccess?: () => void; - onBuildUpdate?: (data: VertexBuildTypeAPI, status: BuildStatus) => void; // Replace any with the actual type if it's not any + onBuildUpdate?: (data: VertexBuildTypeAPI, status: BuildStatus,buildId:string) => void; // Replace any with the actual type if it's not any onBuildComplete?: (allNodesValid: boolean) => void; onBuildError?: (title, list, idList: string[]) => void; onBuildStart?: (idList: string[]) => void; @@ -48,7 +48,7 @@ export async function buildVertices({ let orderResponse; try { orderResponse = await getVerticesOrder(flowId, nodeId); - } catch (error) { + } catch (error:any) { console.log(error); setErrorData({ title: "Oops! Looks like you missed something", @@ -59,6 +59,7 @@ export async function buildVertices({ } if (onGetOrderSuccess) onGetOrderSuccess(); let verticesOrder: Array> = orderResponse.data.ids; + const runId = orderResponse.data.run_id; let vertices_layers: Array> = []; let stop = false; if (validateNodes) { @@ -102,14 +103,14 @@ export async function buildVertices({ onBuildUpdate ) { // If it is, skip building and set the state to inactive - onBuildUpdate(getInactiveVertexData(id), BuildStatus.INACTIVE); + onBuildUpdate(getInactiveVertexData(id), BuildStatus.INACTIVE,runId); buildResults.push(false); continue; } await buildVertex({ flowId, id, - onBuildUpdate, + onBuildUpdate:(data: VertexBuildTypeAPI, status: BuildStatus) => {if(onBuildUpdate) onBuildUpdate(data, status,runId)}, onBuildError, verticesIds, buildResults, diff --git a/src/frontend/src/utils/utils.ts b/src/frontend/src/utils/utils.ts index 77e668c26..d6ebff378 100644 --- a/src/frontend/src/utils/utils.ts +++ b/src/frontend/src/utils/utils.ts @@ -13,7 +13,7 @@ import { tweakType, } from "../types/components"; import { FlowType, NodeType } from "../types/flow"; -import { FlowState, FlowsState } from "../types/tabs"; +import { FlowState } from "../types/tabs"; import { buildTweaks } from "./reactflowUtils"; export function classNames(...classes: Array): string { @@ -217,12 +217,8 @@ export function groupByFamily( })); } -export function buildInputs(flowState?: FlowState): string { - return flowState && - flowState.input_keys && - Object.keys(flowState.input_keys!).length > 0 - ? JSON.stringify(flowState.input_keys) - : '{"input": "message"}'; +export function buildInputs(): string { + return '{"input_value": "message"}'; } export function getRandomElement(array: T[]): T { @@ -323,7 +319,7 @@ export function getPythonApiCode( // node.data.id // } const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `import requests from typing import Optional @@ -382,12 +378,10 @@ export function getCurlCode( ): string { const flowId = flow.id; const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `curl -X POST \\ - ${window.location.protocol}//${ - window.location.host - }/api/v1/process/${flowId} \\ + ${window.location.protocol}//${window.location.host}/api/v1/run/${flowId} \\ -H 'Content-Type: application/json'\\${ !isAuth ? `\n -H 'x-api-key: '\\` : "" } @@ -410,7 +404,7 @@ export function getPythonCode( ): string { const flowName = flow.name; const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `from langflow import load_flow_from_json TWEAKS = ${ tweak && tweak.length > 0 @@ -435,7 +429,7 @@ export function getWidgetCode( ): string { const flowId = flow.id; const flowName = flow.name; - const inputs = buildInputs(flowState); + const inputs = buildInputs(); let chat_input_field = getChatInputField(flow, flowState); return `