jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Merge branch 'zustand/io/migration' of github.com:logspace-ai/langflow into zustand/io/migration

Browse source
This commit is contained in:
igorrCarvalho 2024-02-27 21:05:15 -03:00
commit 9af9647fb3
37 changed files with 458 additions and 255 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.vscode/launch.json vendored
View file

@ -17,6 +17,9 @@
],
"jinja": true,
"justMyCode": true,
"env": {
"LANGFLOW_LOG_LEVEL": "debug"
},
"envFile": "${workspaceFolder}/.env"
},
{

28
src/backend/langflow/api/v1/chat.py
View file

@ -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:

25
src/backend/langflow/api/v1/endpoints.py
View file

@ -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):

16
src/backend/langflow/api/v1/schemas.py
View file

@ -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."""

2
src/backend/langflow/components/chains/ConversationChain.py
View file

@ -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):

4
src/backend/langflow/components/io/ChatOutput.py
View file

@ -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."

2
src/backend/langflow/components/models/AmazonBedrockModel.py
View file

@ -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

2
src/backend/langflow/components/models/AnthropicModel.py
View file

@ -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

2
src/backend/langflow/components/models/AzureOpenAIModel.py
View file

@ -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

2
src/backend/langflow/components/models/BaiduQianfanChatModel.py
View file

@ -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

2
src/backend/langflow/components/models/CTransformersModel.py
View file

@ -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

2
src/backend/langflow/components/models/CohereModel.py
View file

@ -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

2
src/backend/langflow/components/models/GoogleGenerativeAIModel.py
View file

@ -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

2
src/backend/langflow/components/models/HuggingFaceModel.py
View file

@ -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

2
src/backend/langflow/components/models/LlamaCppModel.py
View file

@ -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

2
src/backend/langflow/components/models/OllamaModel.py
View file

@ -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

2
src/backend/langflow/components/models/OpenAIModel.py
View file

@ -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

2
src/backend/langflow/components/models/VertexAiModel.py
View file

@ -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

49
src/backend/langflow/graph/edge/base.py
View file

@ -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)

40
src/backend/langflow/graph/graph/base.py
View file

@ -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

2
src/backend/langflow/graph/schema.py
View file

@ -35,3 +35,5 @@ OUTPUT_COMPONENTS = [
InterfaceComponentTypes.ChatOutput,
InterfaceComponentTypes.TextOutput,
]
INPUT_FIELD_NAME = "input_value"

17
src/backend/langflow/graph/vertex/base.py
View file

@ -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()

24
src/backend/langflow/graph/vertex/types.py
View file

@ -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):

70
src/backend/langflow/interface/custom/code_parser/code_parser.py
View file

@ -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.

67
src/backend/langflow/interface/custom/utils.py
View file

@ -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

39
src/backend/langflow/io/schema.py
View file

@ -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,

4
src/backend/langflow/processing/process.py
View file

@ -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))

5
src/backend/langflow/utils/logger.py
View file

@ -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"

58
src/backend/langflow/utils/validate.py
View file

@ -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=[]), "<string>", "exec")
code_obj = compile(
ast.Module(body=[node], type_ignores=[]), "<string>", "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=[]), "<string>", "exec")
code_obj = compile(
ast.Module(body=[function_code], type_ignores=[]), "<string>", "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=[]), "<string>", "exec")
code_obj = compile(
ast.Module(body=[function_code], type_ignores=[]), "<string>", "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=[]), "<string>", "exec")
code_obj = compile(
ast.Module(body=[class_code], type_ignores=[]), "<string>", "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

107
src/frontend/src/components/newChatView/chatMessage/index.tsx
View file

@ -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<EventSource | undefined>(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<boolean>((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={() => { }}
/>
) : (
<code className={className} {...props}>
@ -264,33 +279,33 @@ dark:prose-invert"
<span className="prose text-primary word-break-break-word dark:prose-invert">
{promptOpen
? template?.split("\n")?.map((line, index) => {
const regex = /{([^}]+)}/g;
let match;
let parts: Array<JSX.Element | string> = [];
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(
<span className="chat-message-highlight">
{chat.message[match[1]]}
</span>
);
}
const regex = /{([^}]+)}/g;
let match;
let parts: Array<JSX.Element | string> = [];
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(
<span className="chat-message-highlight">
{chat.message[match[1]]}
</span>
);
}
// Update last index
lastIndex = regex.lastIndex;
}
// Push text after the last match
if (lastIndex !== line.length) {
parts.push(line.substring(lastIndex));
}
return <p>{parts}</p>;
})
// Update last index
lastIndex = regex.lastIndex;
}
// Push text after the last match
if (lastIndex !== line.length) {
parts.push(line.substring(lastIndex));
}
return <p>{parts}</p>;
})
: chatMessage}
</span>
</>

47
src/frontend/src/components/newChatView/index.tsx
View file

@ -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) => (
<ChatMessage
setLockChat={setLockChat}
lockChat={lockChat}
chat={chat}
lastMessage={chatHistory.length - 1 === index ? true : false}
key={`${chat.id}-${index}`}
key={`${chat.componentId}-${index}`}
updateChat={updateChat}
/>
))

31
src/frontend/src/stores/flowStore.ts
View file

@ -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<FlowStoreType>((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<FlowStoreType>((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<FlowStoreType>((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({

2
src/frontend/src/types/chat/index.ts
View file

@ -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;
};

7
src/frontend/src/types/components/index.ts
View file

@ -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 = {

11
src/frontend/src/types/zustand/flow/index.ts
View file

@ -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;
};

9
src/frontend/src/utils/buildUtils.ts
View file

@ -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<Array<string>> = orderResponse.data.ids;
const runId = orderResponse.data.run_id;
let vertices_layers: Array<Array<string>> = [];
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,

22
src/frontend/src/utils/utils.ts
View file

@ -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>): 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<T>(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: <your 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 `<script src="https://cdn.jsdelivr.net/gh/logspace-ai/langflow-embedded-chat@main/dist/build/static/js/bundle.min.js"></script>