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

Fix formatting and icon naming conventions

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-02-28 21:50:10 -03:00
commit f8fe58bd08
72 changed files with 274 additions and 888 deletions
Download patch file Download diff file Expand all files Collapse all files

32
src/backend/langflow/api/utils.py
View file

@ -20,9 +20,7 @@ API_WORDS = ["api", "key", "token"]
def has_api_terms(word: str):
return "api" in word and (
"key" in word or ("token" in word and "tokens" not in word)
)
return "api" in word and ("key" in word or ("token" in word and "tokens" not in word))
def remove_api_keys(flow: dict):
@ -32,11 +30,7 @@ def remove_api_keys(flow: dict):
node_data = node.get("data").get("node")
template = node_data.get("template")
for value in template.values():
if (
isinstance(value, dict)
and has_api_terms(value["name"])
and value.get("password")
):
if isinstance(value, dict) and has_api_terms(value["name"]) and value.get("password"):
value["value"] = None
return flow
@ -57,9 +51,7 @@ def build_input_keys_response(langchain_object, artifacts):
input_keys_response["input_keys"][key] = value
# If the object has memory, that memory will have a memory_variables attribute
# memory variables should be removed from the input keys
if hasattr(langchain_object, "memory") and hasattr(
langchain_object.memory, "memory_variables"
):
if hasattr(langchain_object, "memory") and hasattr(langchain_object.memory, "memory_variables"):
# Remove memory variables from input keys
input_keys_response["input_keys"] = {
key: value
@ -69,9 +61,7 @@ def build_input_keys_response(langchain_object, artifacts):
# Add memory variables to memory_keys
input_keys_response["memory_keys"] = langchain_object.memory.memory_variables
if hasattr(langchain_object, "prompt") and hasattr(
langchain_object.prompt, "template"
):
if hasattr(langchain_object, "prompt") and hasattr(langchain_object.prompt, "template"):
input_keys_response["template"] = langchain_object.prompt.template
return input_keys_response
@ -106,11 +96,7 @@ def raw_frontend_data_is_valid(raw_frontend_data):
def is_valid_data(frontend_node, raw_frontend_data):
"""Check if the data is valid for processing."""
return (
frontend_node
and "template" in frontend_node
and raw_frontend_data_is_valid(raw_frontend_data)
)
return frontend_node and "template" in frontend_node and raw_frontend_data_is_valid(raw_frontend_data)
def update_template_values(frontend_template, raw_template):
@ -150,9 +136,7 @@ def get_file_path_value(file_path):
# If the path is not in the cache dir, return empty string
# This is to prevent access to files outside the cache dir
# If the path is not a file, return empty string
if not path.exists() or not str(path).startswith(
user_cache_dir("langflow", "langflow")
):
if not path.exists() or not str(path).startswith(user_cache_dir("langflow", "langflow")):
return ""
return file_path
@ -183,9 +167,7 @@ async def check_langflow_version(component: StoreComponentCreate):
langflow_version = get_lf_version_from_pypi()
if langflow_version is None:
raise HTTPException(
status_code=500, detail="Unable to verify the latest version of Langflow"
)
raise HTTPException(status_code=500, detail="Unable to verify the latest version of Langflow")
elif langflow_version != component.last_tested_version:
warnings.warn(
f"Your version of Langflow ({component.last_tested_version}) is outdated. "

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

@ -28,9 +28,7 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
resp = ChatResponse(message=token, type="stream", intermediate_steps="")
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())
async def on_tool_start(
self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
) -> Any:
async def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> Any:
"""Run when tool starts running."""
resp = ChatResponse(
message="",
@ -68,9 +66,7 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
try:
# This is to emulate the stream of tokens
for resp in resps:
await self.socketio_service.emit_token(
to=self.sid, data=resp.model_dump()
)
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())
except Exception as exc:
logger.error(f"Error sending response: {exc}")
@ -96,9 +92,7 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
resp = PromptResponse(
prompt=text,
)
await self.socketio_service.emit_message(
to=self.sid, data=resp.model_dump()
)
await self.socketio_service.emit_message(to=self.sid, data=resp.model_dump())
async def on_agent_action(self, action: AgentAction, **kwargs: Any):
log = f"Thought: {action.log}"
@ -108,9 +102,7 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
logs = log.split("\n")
for log in logs:
resp = ChatResponse(message="", type="stream", intermediate_steps=log)
await self.socketio_service.emit_token(
to=self.sid, data=resp.model_dump()
)
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())
else:
resp = ChatResponse(message="", type="stream", intermediate_steps=log)
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())

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

@ -98,12 +98,8 @@ async def build_vertex(
cache = chat_service.get_cache(flow_id)
if not cache:
# If there's no cache
logger.warning(
f"No cache found for {flow_id}. Building graph starting at {vertex_id}"
)
graph = build_and_cache_graph(
flow_id=flow_id, session=next(get_session()), chat_service=chat_service
)
logger.warning(f"No cache found for {flow_id}. Building graph starting at {vertex_id}")
graph = build_and_cache_graph(flow_id=flow_id, session=next(get_session()), chat_service=chat_service)
else:
graph = cache.get("result")
result_data_response = ResultDataResponse(results={})
@ -195,9 +191,7 @@ async def build_vertex_stream(
else:
graph = cache.get("result")
else:
session_data = await session_service.load_session(
session_id, flow_id=flow_id
)
session_data = await session_service.load_session(session_id, flow_id=flow_id)
graph, artifacts = session_data if session_data else (None, None)
if not graph:
raise ValueError(f"No graph found for {flow_id}.")

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

@ -49,9 +49,7 @@ def get_all(
raise HTTPException(status_code=500, detail=str(exc)) from exc
@router.post(
"/run/{flow_id}", response_model=RunResponse, response_model_exclude_none=True
)
@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,
@ -69,9 +67,7 @@ async def run_flow_with_caching(
input_values_dict = {}
if session_id:
session_data = await session_service.load_session(
session_id, flow_id=flow_id
)
session_data = await session_service.load_session(session_id, flow_id=flow_id)
graph, artifacts = session_data if session_data else (None, None)
task_result: Any = None
if not graph:
@ -89,11 +85,7 @@ async def run_flow_with_caching(
else:
# Get the flow that matches the flow_id and belongs to the user
# flow = session.query(Flow).filter(Flow.id == flow_id).filter(Flow.user_id == api_key_user.id).first()
flow = session.exec(
select(Flow)
.where(Flow.id == flow_id)
.where(Flow.user_id == api_key_user.id)
).first()
flow = session.exec(select(Flow).where(Flow.id == flow_id).where(Flow.user_id == api_key_user.id)).first()
if flow is None:
raise ValueError(f"Flow {flow_id} not found")
@ -116,18 +108,12 @@ async def run_flow_with_caching(
# StatementError('(builtins.ValueError) badly formed hexadecimal UUID string')
if "badly formed hexadecimal UUID string" in str(exc):
# This means the Flow ID is not a valid UUID which means it can't find the flow
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except ValueError as exc:
if f"Flow {flow_id} not found" in str(exc):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
else:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)
) from exc
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
@router.post(
@ -156,8 +142,7 @@ async def process(
"""
# Raise a depreciation warning
logger.warning(
"The /process endpoint is deprecated and will be removed in a future version. "
"Please use /run instead."
"The /process endpoint is deprecated and will be removed in a future version. " "Please use /run instead."
)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
@ -229,16 +214,12 @@ async def custom_component(
built_frontend_node = build_custom_component_template(component, user_id=user.id)
built_frontend_node = update_frontend_node_with_template_values(
built_frontend_node, raw_code.frontend_node
)
built_frontend_node = update_frontend_node_with_template_values(built_frontend_node, raw_code.frontend_node)
return built_frontend_node
@router.post("/custom_component/reload", status_code=HTTPStatus.OK)
async def reload_custom_component(
path: str, user: User = Depends(get_current_active_user)
):
async def reload_custom_component(path: str, user: User = Depends(get_current_active_user)):
from langflow.interface.custom.utils import build_custom_component_template
try:
@ -260,8 +241,6 @@ async def custom_component_update(
):
component = CustomComponent(code=raw_code.code)
component_node = build_custom_component_template(
component, user_id=user.id, update_field=raw_code.field
)
component_node = build_custom_component_template(component, user_id=user.id, update_field=raw_code.field)
# Update the field
return component_node

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

@ -21,26 +21,6 @@ class BuildStatus(Enum):
IN_PROGRESS = "in_progress"
class GraphData(BaseModel):
"""Data inside the exported flow."""
nodes: List[Dict[str, Any]]
edges: List[Dict[str, Any]]
class ExportedFlow(BaseModel):
"""Exported flow from Langflow."""
description: str
name: str
id: str
data: GraphData
class InputRequest(BaseModel):
input: dict
class TweaksRequest(BaseModel):
tweaks: Optional[Dict[str, Dict[str, str]]] = Field(default_factory=dict)
@ -178,9 +158,7 @@ class StreamData(BaseModel):
data: dict
def __str__(self) -> str:
return (
f"event: {self.event}\ndata: {orjson_dumps(self.data, indent_2=False)}\n\n"
)
return f"event: {self.event}\ndata: {orjson_dumps(self.data, indent_2=False)}\n\n"
class CustomComponentCode(BaseModel):

2
src/backend/langflow/components/agents/OpenAIConversationalAgent.py
View file

@ -17,7 +17,7 @@ class ConversationalAgent(CustomComponent):
display_name: str = "OpenAI Conversational Agent"
description: str = "Conversational Agent that can use OpenAI's function calling API"
icon = "OpenAI"
def build_config(self):
openai_function_models = [
"gpt-4-turbo-preview",

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

@ -36,10 +36,8 @@ class ConversationChainComponent(CustomComponent):
# We need to check if it is a string or a BaseMessage
result_str: Text = ""
if hasattr(result, "content") and isinstance(result.content, str):
result_str = result.content
elif isinstance(result, str):
result_str = result
else:
# is dict

5
src/backend/langflow/components/chains/LLMCheckerChain.py
View file

@ -7,9 +7,7 @@ from langflow.field_typing import BaseLanguageModel, Text
class LLMCheckerChainComponent(CustomComponent):
display_name = "LLMCheckerChain"
description = ""
documentation = (
"https://python.langchain.com/docs/modules/chains/additional/llm_checker"
)
documentation = "https://python.langchain.com/docs/modules/chains/additional/llm_checker"
def build_config(self):
return {
@ -21,7 +19,6 @@ class LLMCheckerChainComponent(CustomComponent):
input_value: Text,
llm: BaseLanguageModel,
) -> Text:
chain = LLMCheckerChain.from_llm(llm=llm)
response = chain.invoke({chain.input_key: input_value})
result = response.get(chain.output_key, "")

4
src/backend/langflow/components/chains/LLMMathChain.py
View file

@ -9,9 +9,7 @@ from langflow.field_typing import BaseLanguageModel, BaseMemory, Text
class LLMMathChainComponent(CustomComponent):
display_name = "LLMMathChain"
description = "Chain that interprets a prompt and executes python code to do math."
documentation = (
"https://python.langchain.com/docs/modules/chains/additional/llm_math"
)
documentation = "https://python.langchain.com/docs/modules/chains/additional/llm_math"
def build_config(self):
return {

17
src/backend/langflow/components/chains/SQLGenerator.py
View file

@ -47,19 +47,10 @@ class SQLGeneratorComponent(CustomComponent):
sql_query_chain = create_sql_query_chain(llm=llm, db=db, k=top_k)
else:
# Check if {question} is in the prompt
if (
"{question}" not in prompt_template.template
or "question" not in prompt_template.input_variables
):
raise ValueError(
"Prompt must contain `{question}` to be used with Natural Language to SQL."
)
sql_query_chain = create_sql_query_chain(
llm=llm, db=db, prompt=prompt_template, k=top_k
)
query_writer: Runnable = sql_query_chain | {
"query": lambda x: x.replace("SQLQuery:", "").strip()
}
if "{question}" not in prompt_template.template or "question" not in prompt_template.input_variables:
raise ValueError("Prompt must contain `{question}` to be used with Natural Language to SQL.")
sql_query_chain = create_sql_query_chain(llm=llm, db=db, prompt=prompt_template, k=top_k)
query_writer: Runnable = sql_query_chain | {"query": lambda x: x.replace("SQLQuery:", "").strip()}
response = query_writer.invoke({"question": input_value})
query = response.get("query")
self.status = query

32
src/backend/langflow/components/documentloaders/GatherRecords.py
View file

@ -70,17 +70,11 @@ class GatherRecordsComponent(CustomComponent):
glob = "**/*" if recursive else "*"
paths = walk_level(path_obj, depth) if depth else path_obj.glob(glob)
file_paths = [
Text(p)
for p in paths
if p.is_file() and match_types(p) and is_not_hidden(p)
]
file_paths = [Text(p) for p in paths if p.is_file() and match_types(p) and is_not_hidden(p)]
return file_paths
def parse_file_to_record(
self, file_path: str, silent_errors: bool
) -> Optional[Record]:
def parse_file_to_record(self, file_path: str, silent_errors: bool) -> Optional[Record]:
# Use the partition function to load the file
from unstructured.partition.auto import partition # type: ignore
@ -106,14 +100,9 @@ class GatherRecordsComponent(CustomComponent):
use_multithreading: bool,
) -> List[Optional[Record]]:
if use_multithreading:
records = self.parallel_load_records(
file_paths, silent_errors, max_concurrency
)
records = self.parallel_load_records(file_paths, silent_errors, max_concurrency)
else:
records = [
self.parse_file_to_record(file_path, silent_errors)
for file_path in file_paths
]
records = [self.parse_file_to_record(file_path, silent_errors) for file_path in file_paths]
records = list(filter(None, records))
return records
@ -142,20 +131,13 @@ class GatherRecordsComponent(CustomComponent):
if types is None:
types = []
resolved_path = self.resolve_path(path)
file_paths = self.retrieve_file_paths(
resolved_path, types, load_hidden, recursive, depth
)
file_paths = self.retrieve_file_paths(resolved_path, types, load_hidden, recursive, depth)
loaded_records = []
if use_multithreading:
loaded_records = self.parallel_load_records(
file_paths, silent_errors, max_concurrency
)
loaded_records = self.parallel_load_records(file_paths, silent_errors, max_concurrency)
else:
loaded_records = [
self.parse_file_to_record(file_path, silent_errors)
for file_path in file_paths
]
loaded_records = [self.parse_file_to_record(file_path, silent_errors) for file_path in file_paths]
loaded_records = list(filter(None, loaded_records))
self.status = loaded_records
return loaded_records

2
src/backend/langflow/components/embeddings/HuggingFaceEmbeddings.py
View file

@ -9,7 +9,7 @@ class HuggingFaceEmbeddingsComponent(CustomComponent):
documentation = (
"https://python.langchain.com/docs/modules/data_connection/text_embedding/integrations/sentence_transformers"
)
icon="HuggingFace"
icon = "HuggingFace"
def build_config(self):
return {

3
src/backend/langflow/components/embeddings/HuggingFaceInferenceAPIEmbeddings.py
View file

@ -9,8 +9,7 @@ class HuggingFaceInferenceAPIEmbeddingsComponent(CustomComponent):
display_name = "HuggingFaceInferenceAPIEmbeddings"
description = "HuggingFace sentence_transformers embedding models, API version."
documentation = "https://github.com/huggingface/text-embeddings-inference"
icon="HuggingFace"
icon = "HuggingFace"
def build_config(self):
return {

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

@ -54,9 +54,7 @@ class StoreMessages(CustomComponent):
if not records:
records = []
if not session_id or not sender or not sender_name:
raise ValueError(
"If passing texts, session_id, sender, and sender_name must be provided."
)
raise ValueError("If passing texts, session_id, sender, and sender_name must be provided.")
for text in texts:
record = Record(
text=text,

4
src/backend/langflow/components/io/base/chat.py
View file

@ -45,9 +45,7 @@ class ChatComponent(CustomComponent):
return []
if not session_id or not sender or not sender_name:
raise ValueError(
"All of session_id, sender, and sender_name must be provided."
)
raise ValueError("All of session_id, sender, and sender_name must be provided.")
if isinstance(message, Record):
record = message
record.data.update(

1
src/backend/langflow/components/model_specs/AmazonBedrockSpecs.py
View file

@ -12,7 +12,6 @@ class AmazonBedrockComponent(CustomComponent):
description: str = "LLM model from Amazon Bedrock."
icon = "Amazon"
def build_config(self):
return {
"model_id": {

2
src/backend/langflow/components/model_specs/AnthropicLLMSpecs.py
View file

@ -10,7 +10,7 @@ from langflow import CustomComponent
class AnthropicLLM(CustomComponent):
display_name: str = "AnthropicLLM"
description: str = "Anthropic Chat&Completion large language models."
icon ="Anthropic"
icon = "Anthropic"
def build_config(self):
return {

3
src/backend/langflow/components/model_specs/AnthropicSpecs.py
View file

@ -10,8 +10,7 @@ from langflow.field_typing import BaseLanguageModel, NestedDict
class AnthropicComponent(CustomComponent):
display_name = "Anthropic"
description = "Anthropic large language models."
icon ="Anthropic"
icon = "Anthropic"
def build_config(self):
return {

2
src/backend/langflow/components/model_specs/ChatAnthropicSpecs.py
View file

@ -9,7 +9,7 @@ class ChatAnthropicComponent(CustomComponent):
display_name = "ChatAnthropic"
description = "`Anthropic` chat large language models."
documentation = "https://python.langchain.com/docs/modules/model_io/models/chat/integrations/anthropic"
icon ="Anthropic"
icon = "Anthropic"
def build_config(self):
return {

3
src/backend/langflow/components/model_specs/ChatLiteLLMSpecs.py
View file

@ -127,8 +127,7 @@ class ChatLiteLLMComponent(CustomComponent):
litellm.set_verbose = verbose
except ImportError:
raise ChatLiteLLMException(
"Could not import litellm python package. "
"Please install it with `pip install litellm`"
"Could not import litellm python package. " "Please install it with `pip install litellm`"
)
provider_map = {
"OpenAI": "openai_api_key",

3
src/backend/langflow/components/model_specs/ChatVertexAISpecs.py
View file

@ -10,8 +10,7 @@ from langflow.field_typing import BaseLanguageModel
class ChatVertexAIComponent(CustomComponent):
display_name = "ChatVertexAI"
description = "`Vertex AI` Chat large language models API."
icon="VertexAI"
icon = "VertexAI"
def build_config(self):
return {

3
src/backend/langflow/components/model_specs/HuggingFaceEndpointsSpecs.py
View file

@ -8,8 +8,7 @@ from langflow import CustomComponent
class HuggingFaceEndpointsComponent(CustomComponent):
display_name: str = "Hugging Face Inference API"
description: str = "LLM model from Hugging Face Inference API."
icon="HuggingFace"
icon = "HuggingFace"
def build_config(self):
return {

2
src/backend/langflow/components/model_specs/VertexAISpecs.py
View file

@ -7,7 +7,7 @@ from langchain_community.llms.vertexai import VertexAI
class VertexAIComponent(CustomComponent):
display_name = "VertexAI"
description = "Google Vertex AI large language models"
icon="VertexAI"
icon = "VertexAI"
def build_config(self):
return {

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

@ -9,9 +9,7 @@ from langflow.field_typing import Text
class AnthropicLLM(LCModelComponent):
display_name: str = "AnthropicModel"
description: str = (
"Generate text using Anthropic Chat&Completion large language models."
)
description: str = "Generate text using Anthropic Chat&Completion large language models."
icon = "Anthropic"
def build_config(self):
@ -74,9 +72,7 @@ class AnthropicLLM(LCModelComponent):
try:
output = ChatAnthropic(
model_name=model,
anthropic_api_key=(
SecretStr(anthropic_api_key) if anthropic_api_key else None
),
anthropic_api_key=(SecretStr(anthropic_api_key) if anthropic_api_key else None),
max_tokens_to_sample=max_tokens, # type: ignore
temperature=temperature,
anthropic_api_url=api_endpoint,

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

@ -11,9 +11,7 @@ from langflow.field_typing import Text
class AzureChatOpenAIComponent(LCModelComponent):
display_name: str = "AzureOpenAI Model"
description: str = "Generate text using LLM model from Azure OpenAI."
documentation: str = (
"https://python.langchain.com/docs/integrations/llms/azure_openai"
)
documentation: str = "https://python.langchain.com/docs/integrations/llms/azure_openai"
beta = False
icon = "Azure"

3
src/backend/langflow/components/retrievers/VectaraSelfQueryRetriver.py
View file

@ -17,8 +17,7 @@ class VectaraSelfQueryRetriverComponent(CustomComponent):
description: str = "Implementation of Vectara Self Query Retriever"
documentation = "https://python.langchain.com/docs/integrations/retrievers/self_query/vectara_self_query"
beta = True
icon="Vectara"
icon = "Vectara"
field_config = {
"code": {"show": True},

4
src/backend/langflow/components/utilities/GetRequest.py
View file

@ -32,9 +32,7 @@ class GetRequest(CustomComponent):
},
}
def get_document(
self, session: requests.Session, url: str, headers: Optional[dict], timeout: int
) -> Document:
def get_document(self, session: requests.Session, url: str, headers: Optional[dict], timeout: int) -> Document:
try:
response = session.get(url, headers=headers, timeout=int(timeout))
try:

1
src/backend/langflow/components/utilities/IDGenerator.py
View file

@ -1,4 +1,5 @@
import uuid
from typing import Text
from langflow import CustomComponent

8
src/backend/langflow/components/utilities/PostRequest.py
View file

@ -67,16 +67,12 @@ class PostRequest(CustomComponent):
if not isinstance(document, list) and isinstance(document, Document):
documents: list[Document] = [document]
elif isinstance(document, list) and all(
isinstance(doc, Document) for doc in document
):
elif isinstance(document, list) and all(isinstance(doc, Document) for doc in document):
documents = document
else:
raise ValueError("document must be a Document or a list of Documents")
with requests.Session() as session:
documents = [
self.post_document(session, doc, url, headers) for doc in documents
]
documents = [self.post_document(session, doc, url, headers) for doc in documents]
self.repr_value = documents
return documents

5
src/backend/langflow/components/utilities/RecordsAsText.py
View file

@ -27,10 +27,7 @@ class RecordsAsTextComponent(CustomComponent):
if isinstance(records, Record):
records = [records]
formated_records = [
template.format(text=record.text, data=record.data, **record.data)
for record in records
]
formated_records = [template.format(text=record.text, data=record.data, **record.data) for record in records]
result_string = "\n".join(formated_records)
self.status = result_string
return result_string

4
src/backend/langflow/components/utilities/ShouldRunNext.py
View file

@ -42,9 +42,7 @@ class ShouldRunNext(CustomComponent):
result = result.get("response")
if result.lower() not in ["true", "false"]:
raise ValueError(
"The prompt should generate a boolean response (True or False)."
)
raise ValueError("The prompt should generate a boolean response (True or False).")
# The string should be the words true or false
# if not raise an error
bool_result = result.lower() == "true"

13
src/backend/langflow/components/utilities/UpdateRequest.py
View file

@ -41,9 +41,7 @@ class UpdateRequest(CustomComponent):
) -> Document:
try:
if method == "PATCH":
response = session.patch(
url, headers=headers, data=document.page_content
)
response = session.patch(url, headers=headers, data=document.page_content)
elif method == "PUT":
response = session.put(url, headers=headers, data=document.page_content)
else:
@ -80,17 +78,12 @@ class UpdateRequest(CustomComponent):
if not isinstance(document, list) and isinstance(document, Document):
documents: list[Document] = [document]
elif isinstance(document, list) and all(
isinstance(doc, Document) for doc in document
):
elif isinstance(document, list) and all(isinstance(doc, Document) for doc in document):
documents = document
else:
raise ValueError("document must be a Document or a list of Documents")
with requests.Session() as session:
documents = [
self.update_document(session, doc, url, headers, method)
for doc in documents
]
documents = [self.update_document(session, doc, url, headers, method) for doc in documents]
self.repr_value = documents
return documents

7
src/backend/langflow/components/vectorstores/Chroma.py
View file

@ -84,8 +84,7 @@ class ChromaComponent(CustomComponent):
if chroma_server_host is not None:
chroma_settings = chromadb.config.Settings(
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins
or None,
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins or None,
chroma_server_host=chroma_server_host,
chroma_server_port=chroma_server_port or None,
chroma_server_grpc_port=chroma_server_grpc_port or None,
@ -100,9 +99,7 @@ class ChromaComponent(CustomComponent):
if documents is not None and embedding is not None:
if len(documents) == 0:
raise ValueError(
"If documents are provided, there must be at least one document."
)
raise ValueError("If documents are provided, there must be at least one document.")
chroma = Chroma.from_documents(
documents=documents, # type: ignore
persist_directory=index_directory,

3
src/backend/langflow/components/vectorstores/ChromaSearch.py
View file

@ -93,8 +93,7 @@ class ChromaSearchComponent(LCVectorStoreComponent):
if chroma_server_host is not None:
chroma_settings = chromadb.config.Settings(
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins
or None,
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins or None,
chroma_server_host=chroma_server_host,
chroma_server_port=chroma_server_port or None,
chroma_server_grpc_port=chroma_server_grpc_port or None,

4
src/backend/langflow/components/vectorstores/FAISSSearch.py
View file

@ -34,9 +34,7 @@ class FAISSSearchComponent(LCVectorStoreComponent):
if not folder_path:
raise ValueError("Folder path is required to save the FAISS index.")
path = self.resolve_path(folder_path)
vector_store = FAISS.load_local(
folder_path=Text(path), embeddings=embedding, index_name=index_name
)
vector_store = FAISS.load_local(folder_path=Text(path), embeddings=embedding, index_name=index_name)
if not vector_store:
raise ValueError("Failed to load the FAISS index.")

10
src/backend/langflow/components/vectorstores/MongoDBAtlasVector.py
View file

@ -8,10 +8,8 @@ from langflow.field_typing import Document, Embeddings, NestedDict
class MongoDBAtlasComponent(CustomComponent):
display_name = "MongoDB Atlas"
description = (
"Construct a `MongoDB Atlas Vector Search` vector store from raw documents."
)
icon="MongoDB"
description = "Construct a `MongoDB Atlas Vector Search` vector store from raw documents."
icon = "MongoDB"
def build_config(self):
return {
@ -38,9 +36,7 @@ class MongoDBAtlasComponent(CustomComponent):
try:
from pymongo import MongoClient
except ImportError:
raise ImportError(
"Please install pymongo to use MongoDB Atlas Vector Store"
)
raise ImportError("Please install pymongo to use MongoDB Atlas Vector Store")
try:
mongo_client: MongoClient = MongoClient(mongodb_atlas_cluster_uri)
collection = mongo_client[db_name][collection_name]

2
src/backend/langflow/components/vectorstores/Qdrant.py
View file

@ -10,7 +10,7 @@ from langflow.field_typing import Document, Embeddings, NestedDict
class QdrantComponent(CustomComponent):
display_name = "Qdrant"
description = "Construct Qdrant wrapper from a list of texts."
icon="Qdrant"
icon = "Qdrant"
def build_config(self):
return {

4
src/backend/langflow/components/vectorstores/SupabaseVectorStoreSearch.py
View file

@ -38,9 +38,7 @@ class SupabaseSearchComponent(LCVectorStoreComponent):
supabase_url: str = "",
table_name: str = "",
) -> List[Record]:
supabase: Client = create_client(
supabase_url, supabase_key=supabase_service_key
)
supabase: Client = create_client(supabase_url, supabase_key=supabase_service_key)
vector_store = SupabaseVectorStore(
client=supabase,
embedding=embedding,

4
src/backend/langflow/components/vectorstores/Vectara.py
View file

@ -14,9 +14,7 @@ from langflow.field_typing import BaseRetriever, Document
class VectaraComponent(CustomComponent):
display_name: str = "Vectara"
description: str = "Implementation of Vector Store using Vectara"
documentation = (
"https://python.langchain.com/docs/integrations/vectorstores/vectara"
)
documentation = "https://python.langchain.com/docs/integrations/vectorstores/vectara"
beta = True
icon = "Vectara"
field_config = {

4
src/backend/langflow/components/vectorstores/VectaraSearch.py
View file

@ -11,9 +11,7 @@ from langflow.schema import Record
class VectaraSearchComponent(VectaraComponent, LCVectorStoreComponent):
display_name: str = "Vectara Search"
description: str = "Search a Vectara Vector Store for similar documents."
documentation = (
"https://python.langchain.com/docs/integrations/vectorstores/vectara"
)
documentation = "https://python.langchain.com/docs/integrations/vectorstores/vectara"
beta = True
icon = "Vectara"

4
src/backend/langflow/components/vectorstores/Weaviate.py
View file

@ -11,9 +11,7 @@ from langflow import CustomComponent
class WeaviateVectorStoreComponent(CustomComponent):
display_name: str = "Weaviate"
description: str = "Implementation of Vector Store using Weaviate"
documentation = (
"https://python.langchain.com/docs/integrations/vectorstores/weaviate"
)
documentation = "https://python.langchain.com/docs/integrations/vectorstores/weaviate"
beta = True
field_config = {
"url": {"display_name": "Weaviate URL", "value": "http://localhost:8080"},

4
src/backend/langflow/components/vectorstores/WeaviateSearch.py
View file

@ -11,9 +11,7 @@ from langflow.schema import Record
class WeaviateSearchVectorStore(WeaviateVectorStoreComponent, LCVectorStoreComponent):
display_name: str = "Weaviate Search"
description: str = "Search a Weaviate Vector Store for similar documents."
documentation = (
"https://python.langchain.com/docs/integrations/vectorstores/weaviate"
)
documentation = "https://python.langchain.com/docs/integrations/vectorstores/weaviate"
beta = True
icon = "Weaviate"

11
src/backend/langflow/components/vectorstores/base/model.py
View file

@ -10,7 +10,6 @@ from langflow.schema import Record
class LCVectorStoreComponent(CustomComponent):
display_name: str = "LC Vector Store"
description: str = "Search a LC Vector Store for similar documents."
beta: bool = True
@ -37,14 +36,8 @@ class LCVectorStoreComponent(CustomComponent):
"""
docs: List[Document] = []
if (
input_value
and isinstance(input_value, str)
and hasattr(vector_store, "search")
):
docs = vector_store.search(
query=input_value, search_type=search_type.lower()
)
if input_value and isinstance(input_value, str) and hasattr(vector_store, "search"):
docs = vector_store.search(query=input_value, search_type=search_type.lower())
else:
raise ValueError("Invalid inputs provided.")
return docs_to_records(docs)

4
src/backend/langflow/components/vectorstores/pgvectorSearch.py
View file

@ -15,9 +15,7 @@ class PGVectorSearchComponent(PGVectorComponent, LCVectorStoreComponent):
display_name: str = "PGVector Search"
description: str = "Search a PGVector Store for similar documents."
documentation = (
"https://python.langchain.com/docs/integrations/vectorstores/pgvector"
)
documentation = "https://python.langchain.com/docs/integrations/vectorstores/pgvector"
def build_config(self):
"""

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

@ -13,9 +13,7 @@ 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.")
@ -23,9 +21,7 @@ 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.")
@ -54,24 +50,16 @@ 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"]
@ -88,11 +76,7 @@ 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(
@ -103,10 +87,7 @@ 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 (
@ -118,11 +99,7 @@ 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):
@ -179,9 +156,7 @@ 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)

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

@ -75,9 +75,7 @@ class Graph:
if getattr(vertex, attribute):
getattr(self, f"_{attribute}_vertices").append(vertex.id)
async def _run(
self, inputs: Dict[str, str], stream: bool
) -> List[Optional["ResultData"]]:
async def _run(self, inputs: Dict[str, str], stream: bool) -> List[Optional["ResultData"]]:
"""Runs the graph with the given inputs."""
for vertex_id in self._is_input_vertices:
vertex = self.get_vertex(vertex_id)
@ -100,9 +98,7 @@ class Graph:
outputs.append(vertex.result)
return outputs
async def run(
self, inputs: Dict[str, Union[str, list[str]]], stream: bool
) -> List[Optional["ResultData"]]:
async def run(self, inputs: Dict[str, Union[str, list[str]]], stream: bool) -> List[Optional["ResultData"]]:
"""Runs the graph with the given inputs."""
# inputs is {"message": "Hello, world!"}
@ -114,9 +110,7 @@ class Graph:
if not isinstance(inputs_values, list):
inputs_values = [inputs_values]
for input_value in inputs_values:
run_outputs = await self._run(
{INPUT_FIELD_NAME: input_value}, stream=stream
)
run_outputs = await self._run({INPUT_FIELD_NAME: input_value}, stream=stream)
logger.debug(f"Run outputs: {run_outputs}")
outputs.extend(run_outputs)
return outputs
@ -156,9 +150,7 @@ class Graph:
def build_parent_child_map(self):
parent_child_map = defaultdict(list)
for vertex in self.vertices:
parent_child_map[vertex.id] = [
child.id for child in self.get_successors(vertex)
]
parent_child_map[vertex.id] = [child.id for child in self.get_successors(vertex)]
return parent_child_map
def increment_run_count(self):
@ -333,11 +325,7 @@ class Graph:
return
self.vertices.remove(vertex)
self.vertex_map.pop(vertex_id)
self.edges = [
edge
for edge in self.edges
if edge.source_id != vertex_id and edge.target_id != vertex_id
]
self.edges = [edge for edge in self.edges if edge.source_id != vertex_id and edge.target_id != vertex_id]
def _build_vertex_params(self) -> None:
"""Identifies and handles the LLM vertex within the graph."""
@ -358,9 +346,7 @@ class Graph:
return
for vertex in self.vertices:
if not self._validate_vertex(vertex):
raise ValueError(
f"{vertex.display_name} is not connected to any other components"
)
raise ValueError(f"{vertex.display_name} is not connected to any other components")
def _validate_vertex(self, vertex: Vertex) -> bool:
"""Validates a vertex."""
@ -417,9 +403,7 @@ class Graph:
tasks = []
for vertex_id in layer:
vertex = self.get_vertex(vertex_id)
task = asyncio.create_task(
vertex.build(), name=f"layer-{layer_index}-vertex-{vertex_id}"
)
task = asyncio.create_task(vertex.build(), name=f"layer-{layer_index}-vertex-{vertex_id}")
tasks.append(task)
logger.debug(f"Running layer {layer_index} with {len(tasks)} tasks")
await self._execute_tasks(tasks)
@ -458,9 +442,7 @@ class Graph:
def dfs(vertex):
if state[vertex] == 1:
# We have a cycle
raise ValueError(
"Graph contains a cycle, cannot perform topological sort"
)
raise ValueError("Graph contains a cycle, cannot perform topological sort")
if state[vertex] == 0:
state[vertex] = 1
for edge in vertex.edges:
@ -484,17 +466,11 @@ class Graph:
def get_predecessors(self, vertex):
"""Returns the predecessors of a vertex."""
return [
self.get_vertex(source_id)
for source_id in self.predecessor_map.get(vertex.id, [])
]
return [self.get_vertex(source_id) for source_id in self.predecessor_map.get(vertex.id, [])]
def get_successors(self, vertex):
"""Returns the successors of a vertex."""
return [
self.get_vertex(target_id)
for target_id in self.successor_map.get(vertex.id, [])
]
return [self.get_vertex(target_id) for target_id in self.successor_map.get(vertex.id, [])]
def get_vertex_neighbors(self, vertex: Vertex) -> Dict[Vertex, int]:
"""Returns the neighbors of a vertex."""
@ -533,9 +509,7 @@ class Graph:
edges.append(ContractEdge(source, target, edge))
return edges
def _get_vertex_class(
self, node_type: str, node_base_type: str, node_id: str
) -> Type[Vertex]:
def _get_vertex_class(self, node_type: str, node_base_type: str, node_id: str) -> Type[Vertex]:
"""Returns the node class based on the node type."""
# First we check for the node_base_type
node_name = node_id.split("-")[0]
@ -566,18 +540,14 @@ class Graph:
vertex_type: str = vertex_data["type"] # type: ignore
vertex_base_type: str = vertex_data["node"]["template"]["_type"] # type: ignore
VertexClass = self._get_vertex_class(
vertex_type, vertex_base_type, vertex_data["id"]
)
VertexClass = self._get_vertex_class(vertex_type, vertex_base_type, vertex_data["id"])
vertex_instance = VertexClass(vertex, graph=self)
vertex_instance.set_top_level(self.top_level_vertices)
vertices.append(vertex_instance)
return vertices
def get_children_by_vertex_type(
self, vertex: Vertex, vertex_type: str
) -> List[Vertex]:
def get_children_by_vertex_type(self, vertex: Vertex, vertex_type: str) -> List[Vertex]:
"""Returns the children of a vertex based on the vertex type."""
children = []
vertex_types = [vertex.data["type"]]
@ -589,9 +559,7 @@ class Graph:
def __repr__(self):
vertex_ids = [vertex.id for vertex in self.vertices]
edges_repr = "\n".join(
[f"{edge.source_id} --> {edge.target_id}" for edge in self.edges]
)
edges_repr = "\n".join([f"{edge.source_id} --> {edge.target_id}" for edge in self.edges])
return f"Graph:\nNodes: {vertex_ids}\nConnections:\n{edges_repr}"
def sort_up_to_vertex(self, vertex_id: str) -> List[Vertex]:
@ -622,9 +590,7 @@ class Graph:
"""Performs a layered topological sort of the vertices in the graph."""
# Queue for vertices with no incoming edges
queue = deque(
vertex.id for vertex in vertices if self.in_degree_map[vertex.id] == 0
)
queue = deque(vertex.id for vertex in vertices if self.in_degree_map[vertex.id] == 0)
layers: List[List[str]] = []
current_layer = 0
@ -680,9 +646,7 @@ class Graph:
return refined_layers
def sort_chat_inputs_first(
self, vertices_layers: List[List[str]]
) -> List[List[str]]:
def sort_chat_inputs_first(self, vertices_layers: List[List[str]]) -> List[List[str]]:
chat_inputs_first = []
for layer in vertices_layers:
for vertex_id in layer:
@ -711,15 +675,11 @@ class Graph:
self._sorted_vertices_layers = vertices_layers
return vertices_layers
def sort_interface_components_first(
self, vertices_layers: List[List[str]]
) -> List[List[str]]:
def sort_interface_components_first(self, vertices_layers: List[List[str]]) -> List[List[str]]:
"""Sorts the vertices in the graph so that vertices containing ChatInput or ChatOutput come first."""
def contains_interface_component(vertex):
return any(
component.value in vertex for component in InterfaceComponentTypes
)
return any(component.value in vertex for component in InterfaceComponentTypes)
# Sort each inner list so that vertices containing ChatInput or ChatOutput come first
sorted_vertices = [
@ -731,22 +691,16 @@ class Graph:
]
return sorted_vertices
def sort_by_avg_build_time(
self, vertices_layers: List[List[str]]
) -> List[List[str]]:
def sort_by_avg_build_time(self, vertices_layers: List[List[str]]) -> List[List[str]]:
"""Sorts the vertices in the graph so that vertices with the lowest average build time come first."""
def sort_layer_by_avg_build_time(vertices_ids: List[str]) -> List[str]:
"""Sorts the vertices in the graph so that vertices with the lowest average build time come first."""
if len(vertices_ids) == 1:
return vertices_ids
vertices_ids.sort(
key=lambda vertex_id: self.get_vertex(vertex_id).avg_build_time
)
vertices_ids.sort(key=lambda vertex_id: self.get_vertex(vertex_id).avg_build_time)
return vertices_ids
sorted_vertices = [
sort_layer_by_avg_build_time(layer) for layer in vertices_layers
]
sorted_vertices = [sort_layer_by_avg_build_time(layer) for layer in vertices_layers]
return sorted_vertices

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

@ -47,13 +47,8 @@ class Vertex:
self.will_stream = False
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
)
self.is_output = any(
output_component_name in self.id
for output_component_name in OUTPUT_COMPONENTS
)
self.is_input = any(input_component_name in self.id for input_component_name in INPUT_COMPONENTS)
self.is_output = any(output_component_name in self.id for output_component_name in OUTPUT_COMPONENTS)
self.has_session_id = None
self._custom_component = None
self.has_external_input = False
@ -87,17 +82,11 @@ class Vertex:
def set_state(self, state: str):
self.state = VertexStates[state]
if (
self.state == VertexStates.INACTIVE
and self.graph.in_degree_map[self.id] < 2
):
if self.state == VertexStates.INACTIVE and self.graph.in_degree_map[self.id] < 2:
# If the vertex is inactive and has only one in degree
# it means that it is not a merge point in the graph
self.graph.inactive_vertices.add(self.id)
elif (
self.state == VertexStates.ACTIVE
and self.id in self.graph.inactive_vertices
):
elif self.state == VertexStates.ACTIVE and self.id in self.graph.inactive_vertices:
self.graph.inactive_vertices.remove(self.id)
@property
@ -114,9 +103,7 @@ class Vertex:
# If the Vertex.type is a power component
# then we need to return the built object
# instead of the result dict
if self.is_interface_component and not isinstance(
self._built_object, UnbuiltObject
):
if self.is_interface_component and not isinstance(self._built_object, UnbuiltObject):
result = self._built_object
# if it is not a dict or a string and hasattr model_dump then
# return the model_dump
@ -126,11 +113,7 @@ class Vertex:
if isinstance(self._built_result, UnbuiltResult):
return {}
return (
self._built_result
if isinstance(self._built_result, dict)
else {"result": self._built_result}
)
return self._built_result if isinstance(self._built_result, dict) else {"result": self._built_result}
def set_artifacts(self) -> None:
pass
@ -196,31 +179,19 @@ class Vertex:
self.selected_output_type = self.data["node"].get("selected_output_type")
self.is_input = self.data["node"].get("is_input") or self.is_input
self.is_output = self.data["node"].get("is_output") or self.is_output
template_dicts = {
key: value
for key, value in self.data["node"]["template"].items()
if isinstance(value, dict)
}
template_dicts = {key: value for key, value in self.data["node"]["template"].items() if isinstance(value, dict)}
self.has_session_id = "session_id" in template_dicts
self.required_inputs = [
template_dicts[key]["type"]
for key, value in template_dicts.items()
if value["required"]
template_dicts[key]["type"] for key, value in template_dicts.items() if value["required"]
]
self.optional_inputs = [
template_dicts[key]["type"]
for key, value in template_dicts.items()
if not value["required"]
template_dicts[key]["type"] for key, value in template_dicts.items() if not value["required"]
]
# Add the template_dicts[key]["input_types"] to the optional_inputs
self.optional_inputs.extend(
[
input_type
for value in template_dicts.values()
for input_type in value.get("input_types", [])
]
[input_type for value in template_dicts.values() for input_type in value.get("input_types", [])]
)
template_dict = self.data["node"]["template"]
@ -267,11 +238,7 @@ class Vertex:
self.updated_raw_params = False
return
template_dict = {
key: value
for key, value in self.data["node"]["template"].items()
if isinstance(value, dict)
}
template_dict = {key: value for key, value in self.data["node"]["template"].items() if isinstance(value, dict)}
params = {}
for edge in self.edges:
@ -322,11 +289,7 @@ class Vertex:
# list of dicts, so we need to convert it to a dict
# before passing it to the build method
if isinstance(val, list):
params[key] = {
k: v
for item in value.get("value", [])
for k, v in item.items()
}
params[key] = {k: v for item in value.get("value", []) for k, v in item.items()}
elif isinstance(val, dict):
params[key] = val
elif value.get("type") == "int" and val is not None:
@ -419,9 +382,7 @@ class Vertex:
if isinstance(self._built_object, str):
self._built_result = self._built_object
result = await generate_result(
self._built_object, inputs, self.has_external_output, session_id
)
result = await generate_result(self._built_object, inputs, self.has_external_output, session_id)
self._built_result = result
async def _build_each_node_in_params_dict(self, user_id=None):
@ -451,9 +412,7 @@ class Vertex:
"""
return all(self._is_node(node) for node in value)
async def get_result(
self, requester: Optional["Vertex"] = None, user_id=None, timeout=None
) -> Any:
async def get_result(self, requester: Optional["Vertex"] = None, user_id=None, timeout=None) -> Any:
# PLEASE REVIEW THIS IF STATEMENT
# Check if the Vertex was built already
if self._built:
@ -487,9 +446,7 @@ class Vertex:
self._extend_params_list_with_result(key, result)
self.params[key] = result
async def _build_list_of_nodes_and_update_params(
self, key, nodes: List["Vertex"], user_id=None
):
async def _build_list_of_nodes_and_update_params(self, key, nodes: List["Vertex"], user_id=None):
"""
Iterates over a list of nodes, builds each and updates the params dictionary.
"""
@ -532,7 +489,6 @@ class Vertex:
if self.base_type is None:
raise ValueError(f"Base type for node {self.display_name} not found")
try:
result = await loading.instantiate_class(
node_type=self.vertex_type,
base_type=self.base_type,
@ -544,9 +500,7 @@ class Vertex:
except Exception as exc:
logger.exception(exc)
raise ValueError(
f"Error building node {self.display_name}: {str(exc)}"
) from exc
raise ValueError(f"Error building node {self.display_name}: {str(exc)}") from exc
def _update_built_object_and_artifacts(self, result):
"""
@ -626,24 +580,16 @@ class Vertex:
return self._built_object
# Get the requester edge
requester_edge = next(
(edge for edge in self.edges if edge.target_id == requester.id), None
)
requester_edge = next((edge for edge in self.edges if edge.target_id == requester.id), None)
# Return the result of the requester edge
return (
None
if requester_edge is None
else await requester_edge.get_result(source=self, target=requester)
)
return None if requester_edge is None else await requester_edge.get_result(source=self, target=requester)
def add_edge(self, edge: "ContractEdge") -> None:
if edge not in self.edges:
self.edges.append(edge)
def __repr__(self) -> str:
return (
f"Vertex(display_name={self.display_name}, id={self.id}, data={self.data})"
)
return f"Vertex(display_name={self.display_name}, id={self.id}, data={self.data})"
def __eq__(self, __o: object) -> bool:
try:
@ -656,11 +602,7 @@ class Vertex:
def _built_object_repr(self):
# Add a message with an emoji, stars for sucess,
return (
"Built sucessfully ✨"
if self._built_object is not None
else "Failed to build 😵‍💫"
)
return "Built sucessfully ✨" if self._built_object is not None else "Failed to build 😵‍💫"
class StatefulVertex(Vertex):

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

@ -123,11 +123,9 @@ class DocumentLoaderVertex(StatefulVertex):
# show how many documents are in the list?
if not isinstance(self._built_object, UnbuiltObject):
avg_length = sum(
len(doc.page_content)
for doc in self._built_object
if hasattr(doc, "page_content")
) / len(self._built_object)
avg_length = sum(len(doc.page_content) for doc in self._built_object if hasattr(doc, "page_content")) / len(
self._built_object
)
return f"""{self.display_name}({len(self._built_object)} documents)
\nAvg. Document Length (characters): {int(avg_length)}
Documents: {self._built_object[:3]}..."""
@ -200,9 +198,7 @@ class TextSplitterVertex(StatefulVertex):
# show how many documents are in the list?
if not isinstance(self._built_object, UnbuiltObject):
avg_length = sum(len(doc.page_content) for doc in self._built_object) / len(
self._built_object
)
avg_length = sum(len(doc.page_content) for doc in self._built_object) / len(self._built_object)
return f"""{self.vertex_type}({len(self._built_object)} documents)
\nAvg. Document Length (characters): {int(avg_length)}
\nDocuments: {self._built_object[:3]}..."""
@ -249,27 +245,18 @@ class PromptVertex(StatelessVertex):
user_id = kwargs.get("user_id", None)
tools = kwargs.get("tools", [])
if not self._built or force:
if (
"input_variables" not in self.params
or self.params["input_variables"] is None
):
if "input_variables" not in self.params or self.params["input_variables"] is None:
self.params["input_variables"] = []
# Check if it is a ZeroShotPrompt and needs a tool
if "ShotPrompt" in self.vertex_type:
tools = (
[tool_node.build(user_id=user_id) for tool_node in tools]
if tools is not None
else []
)
tools = [tool_node.build(user_id=user_id) for tool_node in tools] if tools is not None else []
# flatten the list of tools if it is a list of lists
# first check if it is a list
if tools and isinstance(tools, list) and isinstance(tools[0], list):
tools = flatten_list(tools)
self.params["tools"] = tools
prompt_params = [
key
for key, value in self.params.items()
if isinstance(value, str) and key != "format_instructions"
key for key, value in self.params.items() if isinstance(value, str) and key != "format_instructions"
]
else:
prompt_params = ["template"]
@ -279,20 +266,14 @@ class PromptVertex(StatelessVertex):
prompt_text = self.params[param]
variables = extract_input_variables_from_prompt(prompt_text)
self.params["input_variables"].extend(variables)
self.params["input_variables"] = list(
set(self.params["input_variables"])
)
self.params["input_variables"] = list(set(self.params["input_variables"]))
elif isinstance(self.params, dict):
self.params.pop("input_variables", None)
await self._build(user_id=user_id)
def _built_object_repr(self):
if (
not self.artifacts
or self._built_object is None
or not hasattr(self._built_object, "format")
):
if not self.artifacts or self._built_object is None or not hasattr(self._built_object, "format"):
return super()._built_object_repr()
elif isinstance(self._built_object, UnbuiltObject):
return super()._built_object_repr()
@ -304,9 +285,7 @@ class PromptVertex(StatelessVertex):
# so the prompt format doesn't break
artifacts.pop("handle_keys", None)
try:
if not hasattr(self._built_object, "template") and hasattr(
self._built_object, "prompt"
):
if not hasattr(self._built_object, "template") and hasattr(self._built_object, "prompt"):
template = self._built_object.prompt.template
else:
template = self._built_object.template
@ -314,11 +293,7 @@ class PromptVertex(StatelessVertex):
if value:
replace_key = "{" + key + "}"
template = template.replace(replace_key, value)
return (
template
if isinstance(template, str)
else f"{self.vertex_type}({template})"
)
return template if isinstance(template, str) else f"{self.vertex_type}({template})"
except KeyError:
return str(self._built_object)
@ -483,7 +458,6 @@ class RoutingVertex(StatelessVertex):
def dict_to_codeblock(d: dict) -> str:
serialized = {key: serialize_field(val) for key, val in d.items()}
json_str = json.dumps(serialized, indent=4)
return f"```json\n{json_str}\n```"

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

@ -95,9 +95,7 @@ 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))
@ -146,9 +144,7 @@ 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)
@ -190,22 +186,14 @@ 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]]:
@ -223,17 +211,11 @@ 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]]:
@ -337,9 +319,7 @@ 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)

33
src/backend/langflow/interface/custom/custom_component/custom_component.py
View file

@ -119,9 +119,7 @@ class CustomComponent(Component):
def tree(self):
return self.get_code_tree(self.code or "")
def to_records(
self, data: Any, text_key: str = "text", data_key: str = "data"
) -> List[Record]:
def to_records(self, data: Any, text_key: str = "text", data_key: str = "data") -> List[Record]:
"""
Convert data into a list of records.
@ -148,9 +146,7 @@ class CustomComponent(Component):
return records
def create_references_from_records(
self, records: List[Record], include_data: bool = False
) -> str:
def create_references_from_records(self, records: List[Record], include_data: bool = False) -> str:
"""
Create references from a list of records.
@ -185,8 +181,7 @@ class CustomComponent(Component):
detail={
"error": "Type hint Error",
"traceback": (
"Prompt type is not supported in the build method."
" Try using PromptTemplate instead."
"Prompt type is not supported in the build method." " Try using PromptTemplate instead."
),
},
)
@ -200,20 +195,14 @@ class CustomComponent(Component):
if not self.code:
return {}
component_classes = [
cls
for cls in self.tree["classes"]
if self.code_class_base_inheritance in cls["bases"]
]
component_classes = [cls for cls in self.tree["classes"] if self.code_class_base_inheritance in cls["bases"]]
if not component_classes:
return {}
# Assume the first Component class is the one we're interested in
component_class = component_classes[0]
build_methods = [
method
for method in component_class["methods"]
if method["name"] == self.function_entrypoint_name
method for method in component_class["methods"] if method["name"] == self.function_entrypoint_name
]
return build_methods[0] if build_methods else {}
@ -270,9 +259,7 @@ class CustomComponent(Component):
# Retrieve and decrypt the credential by name for the current user
db_service = get_db_service()
with session_getter(db_service) as session:
return credential_service.get_credential(
user_id=self._user_id or "", name=name, session=session
)
return credential_service.get_credential(user_id=self._user_id or "", name=name, session=session)
return get_credential
@ -282,9 +269,7 @@ class CustomComponent(Component):
credential_service = get_credential_service()
db_service = get_db_service()
with session_getter(db_service) as session:
return credential_service.list_credentials(
user_id=self._user_id, session=session
)
return credential_service.list_credentials(user_id=self._user_id, session=session)
def index(self, value: int = 0):
"""Returns a function that returns the value at the given index in the iterable."""
@ -328,9 +313,7 @@ class CustomComponent(Component):
get_session = get_session or session_getter
db_service = get_db_service()
with get_session(db_service) as session:
flows = session.exec(
select(Flow).where(Flow.user_id == self._user_id)
).all()
flows = session.exec(select(Flow).where(Flow.user_id == self._user_id)).all()
return flows
except Exception as e:
raise ValueError("Session is invalid") from e

54
src/backend/langflow/interface/custom/directory_reader/directory_reader.py
View file

@ -80,13 +80,9 @@ class DirectoryReader:
except Exception as e:
logger.error(f"Error while loading component: {e}")
continue
items.append(
{"name": menu["name"], "path": menu["path"], "components": components}
)
items.append({"name": menu["name"], "path": menu["path"], "components": components})
filtered = [menu for menu in items if menu["components"]]
logger.debug(
f'Filtered components {"with errors" if with_errors else ""}: {len(filtered)}'
)
logger.debug(f'Filtered components {"with errors" if with_errors else ""}: {len(filtered)}')
return {"menu": filtered}
def validate_code(self, file_content):
@ -119,9 +115,7 @@ class DirectoryReader:
Walk through the directory path and return a list of all .py files.
"""
if not (safe_path := self.get_safe_path()):
raise CustomComponentPathValueError(
f"The path needs to start with '{self.base_path}'."
)
raise CustomComponentPathValueError(f"The path needs to start with '{self.base_path}'.")
file_list = []
safe_path_obj = Path(safe_path)
@ -131,11 +125,7 @@ class DirectoryReader:
# any folders below [folder] will be ignored
# basically the parent folder of the file should be a
# folder in the safe_path
if (
file_path.is_file()
and file_path.parent.parent == safe_path_obj
and not file_path.name.startswith("__")
):
if file_path.is_file() and file_path.parent.parent == safe_path_obj and not file_path.name.startswith("__"):
file_list.append(str(file_path))
return file_list
@ -173,9 +163,7 @@ class DirectoryReader:
for node in ast.walk(module):
if isinstance(node, ast.FunctionDef):
for arg in node.args.args:
if self._is_type_hint_in_arg_annotation(
arg.annotation, type_hint_name
):
if self._is_type_hint_in_arg_annotation(arg.annotation, type_hint_name):
return True
except SyntaxError:
# Returns False if the code is not valid Python
@ -193,16 +181,14 @@ class DirectoryReader:
and annotation.value.id == type_hint_name
)
def is_type_hint_used_but_not_imported(
self, type_hint_name: str, code: str
) -> bool:
def is_type_hint_used_but_not_imported(self, type_hint_name: str, code: str) -> bool:
"""
Check if a type hint is used but not imported in the given code.
"""
try:
return self._is_type_hint_used_in_args(
return self._is_type_hint_used_in_args(type_hint_name, code) and not self._is_type_hint_imported(
type_hint_name, code
) and not self._is_type_hint_imported(type_hint_name, code)
)
except SyntaxError:
# Returns True if there's something wrong with the code
# TODO : Find a better way to handle this
@ -223,9 +209,9 @@ class DirectoryReader:
return False, "Syntax error"
elif not self.validate_build(file_content):
return False, "Missing build function"
elif self._is_type_hint_used_in_args(
elif self._is_type_hint_used_in_args("Optional", file_content) and not self._is_type_hint_imported(
"Optional", file_content
) and not self._is_type_hint_imported("Optional", file_content):
):
return (
False,
"Type hint 'Optional' is used but not imported in the code.",
@ -241,18 +227,14 @@ class DirectoryReader:
from the .py files in the directory.
"""
response = {"menu": []}
logger.debug(
"-------------------- Building component menu list --------------------"
)
logger.debug("-------------------- Building component menu list --------------------")
for file_path in file_paths:
menu_name = os.path.basename(os.path.dirname(file_path))
filename = os.path.basename(file_path)
validation_result, result_content = self.process_file(file_path)
if not validation_result:
logger.error(
f"Error while processing file {file_path}: {result_content}"
)
logger.error(f"Error while processing file {file_path}: {result_content}")
menu_result = self.find_menu(response, menu_name) or {
"name": menu_name,
@ -265,9 +247,7 @@ class DirectoryReader:
# first check if it's already CamelCase
if "_" in component_name:
component_name_camelcase = " ".join(
word.title() for word in component_name.split("_")
)
component_name_camelcase = " ".join(word.title() for word in component_name.split("_"))
else:
component_name_camelcase = component_name
@ -275,9 +255,7 @@ class DirectoryReader:
try:
output_types = self.get_output_types_from_code(result_content)
except Exception as exc:
logger.exception(
f"Error while getting output types from code: {str(exc)}"
)
logger.exception(f"Error while getting output types from code: {str(exc)}")
output_types = [component_name_camelcase]
else:
output_types = [component_name_camelcase]
@ -293,9 +271,7 @@ class DirectoryReader:
if menu_result not in response["menu"]:
response["menu"].append(menu_result)
logger.debug(
"-------------------- Component menu list built --------------------"
)
logger.debug("-------------------- Component menu list built --------------------")
return response
@staticmethod

18
src/backend/langflow/interface/custom/directory_reader/utils.py
View file

@ -8,11 +8,7 @@ from langflow.template.frontend_node.custom_components import (
def merge_nested_dicts_with_renaming(dict1, dict2):
for key, value in dict2.items():
if (
key in dict1
and isinstance(value, dict)
and isinstance(dict1.get(key), dict)
):
if key in dict1 and isinstance(value, dict) and isinstance(dict1.get(key), dict):
for sub_key, sub_value in value.items():
# if sub_key in dict1[key]:
# new_key = get_new_key(dict1[key], sub_key)
@ -69,9 +65,7 @@ def build_custom_component_list_from_path(path: str):
file_list = load_files_from_path(path)
reader = DirectoryReader(path, False)
valid_components, invalid_components = build_and_validate_all_files(
reader, file_list
)
valid_components, invalid_components = build_and_validate_all_files(reader, file_list)
valid_menu = build_valid_menu(valid_components)
invalid_menu = build_invalid_menu(invalid_components)
@ -118,9 +112,7 @@ def build_invalid_menu_items(menu_item):
menu_items[component_name] = component_template
logger.debug(f"Added {component_name} to invalid menu.")
except Exception as exc:
logger.exception(
f"Error while creating custom component [{component_name}]: {str(exc)}"
)
logger.exception(f"Error while creating custom component [{component_name}]: {str(exc)}")
return menu_items
@ -154,7 +146,5 @@ def build_menu_items(menu_item):
menu_items[component_name] = component_template
except Exception as exc:
logger.error(f"Error loading Component: {component['output_types']}")
logger.exception(
f"Error while building custom component {component['output_types']}: {exc}"
)
logger.exception(f"Error while building custom component {component['output_types']}: {exc}")
return menu_items

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

@ -27,18 +27,14 @@ 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(),
},
)
@ -67,18 +63,14 @@ 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(),
},
)
@ -153,14 +145,10 @@ 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", "")
@ -191,9 +179,7 @@ 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,
@ -231,9 +217,7 @@ 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
@ -261,9 +245,7 @@ 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
@ -316,24 +298,16 @@ def build_custom_component_template(
try:
frontend_node = build_frontend_node(custom_component.template_config)
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)
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)
reorder_fields(frontend_node, custom_instance._get_field_order())
@ -344,9 +318,7 @@ 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
@ -370,9 +342,7 @@ 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:
@ -383,9 +353,7 @@ 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
)

58
src/backend/langflow/interface/initialize/loading.py
View file

@ -143,13 +143,9 @@ async def instantiate_based_on_type(
return class_object(**params)
async def instantiate_custom_component(
node_type, class_object, params, user_id, vertex
):
async def instantiate_custom_component(node_type, class_object, params, user_id, vertex):
params_copy = params.copy()
class_object: Type["CustomComponent"] = eval_custom_component_code(
params_copy.pop("code")
)
class_object: Type["CustomComponent"] = eval_custom_component_code(params_copy.pop("code"))
custom_component: "CustomComponent" = class_object(
user_id=user_id,
parameters=params_copy,
@ -223,9 +219,7 @@ def instantiate_memory(node_type, class_object, params):
# I want to catch a specific attribute error that happens
# when the object does not have a cursor attribute
except Exception as exc:
if "object has no attribute 'cursor'" in str(
exc
) or 'object has no field "conn"' in str(exc):
if "object has no attribute 'cursor'" in str(exc) or 'object has no field "conn"' in str(exc):
raise AttributeError(
(
"Failed to build connection to database."
@ -268,9 +262,7 @@ def instantiate_agent(node_type, class_object: Type[agent_module.Agent], params:
if class_method := getattr(class_object, method, None):
agent = class_method(**params)
tools = params.get("tools", [])
return AgentExecutor.from_agent_and_tools(
agent=agent, tools=tools, handle_parsing_errors=True
)
return AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, handle_parsing_errors=True)
return load_agent_executor(class_object, params)
@ -326,11 +318,7 @@ def instantiate_embedding(node_type, class_object, params: Dict):
try:
return class_object(**params)
except ValidationError:
params = {
key: value
for key, value in params.items()
if key in class_object.model_fields
}
params = {key: value for key, value in params.items() if key in class_object.model_fields}
return class_object(**params)
@ -342,9 +330,7 @@ def instantiate_vectorstore(class_object: Type[VectorStore], params: Dict):
if "texts" in params:
params["documents"] = params.pop("texts")
if "documents" in params:
params["documents"] = [
doc for doc in params["documents"] if isinstance(doc, Document)
]
params["documents"] = [doc for doc in params["documents"] if isinstance(doc, Document)]
if initializer := vecstore_initializer.get(class_object.__name__):
vecstore = initializer(class_object, params)
else:
@ -359,9 +345,7 @@ def instantiate_vectorstore(class_object: Type[VectorStore], params: Dict):
return vecstore
def instantiate_documentloader(
node_type: str, class_object: Type[BaseLoader], params: Dict
):
def instantiate_documentloader(node_type: str, class_object: Type[BaseLoader], params: Dict):
if "file_filter" in params:
# file_filter will be a string but we need a function
# that will be used to filter the files using file_filter
@ -370,17 +354,13 @@ def instantiate_documentloader(
# in x and if it is, we will return True
file_filter = params.pop("file_filter")
extensions = file_filter.split(",")
params["file_filter"] = lambda x: any(
extension.strip() in x for extension in extensions
)
params["file_filter"] = lambda x: any(extension.strip() in x for extension in extensions)
metadata = params.pop("metadata", None)
if metadata and isinstance(metadata, str):
try:
metadata = orjson.loads(metadata)
except json.JSONDecodeError as exc:
raise ValueError(
"The metadata you provided is not a valid JSON string."
) from exc
raise ValueError("The metadata you provided is not a valid JSON string.") from exc
if node_type == "WebBaseLoader":
if web_path := params.pop("web_path", None):
@ -413,16 +393,12 @@ def instantiate_textsplitter(
"Try changing the chunk_size of the Text Splitter."
) from exc
if (
"separator_type" in params and params["separator_type"] == "Text"
) or "separator_type" not in params:
if ("separator_type" in params and params["separator_type"] == "Text") or "separator_type" not in params:
params.pop("separator_type", None)
# separators might come in as an escaped string like \\n
# so we need to convert it to a string
if "separators" in params:
params["separators"] = (
params["separators"].encode().decode("unicode-escape")
)
params["separators"] = params["separators"].encode().decode("unicode-escape")
text_splitter = class_object(**params)
else:
from langchain.text_splitter import Language
@ -449,8 +425,7 @@ def replace_zero_shot_prompt_with_prompt_template(nodes):
tools = [
tool
for tool in nodes
if tool["type"] != "chatOutputNode"
and "Tool" in tool["data"]["node"]["base_classes"]
if tool["type"] != "chatOutputNode" and "Tool" in tool["data"]["node"]["base_classes"]
]
node["data"] = build_prompt_template(prompt=node["data"], tools=tools)
break
@ -464,9 +439,7 @@ def load_agent_executor(agent_class: type[agent_module.Agent], params, **kwargs)
# agent has hidden args for memory. might need to be support
# memory = params["memory"]
# if allowed_tools is not a list or set, make it a list
if not isinstance(allowed_tools, (list, set)) and isinstance(
allowed_tools, BaseTool
):
if not isinstance(allowed_tools, (list, set)) and isinstance(allowed_tools, BaseTool):
allowed_tools = [allowed_tools]
tool_names = [tool.name for tool in allowed_tools]
# Agent class requires an output_parser but Agent classes
@ -494,10 +467,7 @@ def build_prompt_template(prompt, tools):
format_instructions = prompt["node"]["template"]["format_instructions"]["value"]
tool_strings = "\n".join(
[
f"{tool['data']['node']['name']}: {tool['data']['node']['description']}"
for tool in tools
]
[f"{tool['data']['node']['name']}: {tool['data']['node']['description']}" for tool in tools]
)
tool_names = ", ".join([tool["data"]["node"]["name"] for tool in tools])
format_instructions = format_instructions.format(tool_names=tool_names)

12
src/backend/langflow/main.py
View file

@ -18,9 +18,7 @@ from langflow.utils.logger import configure
def get_lifespan(fix_migration=False, socketio_server=None):
@asynccontextmanager
async def lifespan(app: FastAPI):
initialize_services(
fix_migration=fix_migration, socketio_server=socketio_server
)
initialize_services(fix_migration=fix_migration, socketio_server=socketio_server)
setup_llm_caching()
LangfuseInstance.update()
yield
@ -33,9 +31,7 @@ def create_app():
"""Create the FastAPI app and include the router."""
configure()
socketio_server = socketio.AsyncServer(
async_mode="asgi", cors_allowed_origins="*", logger=True
)
socketio_server = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*", logger=True)
lifespan = get_lifespan(socketio_server=socketio_server)
app = FastAPI(lifespan=lifespan)
origins = ["*"]
@ -102,9 +98,7 @@ def get_static_files_dir():
return frontend_path / "frontend"
def setup_app(
static_files_dir: Optional[Path] = None, backend_only: bool = False
) -> FastAPI:
def setup_app(static_files_dir: Optional[Path] = None, backend_only: bool = False) -> FastAPI:
"""Setup the FastAPI app."""
# get the directory of the current file
if not static_files_dir:

14
src/backend/langflow/processing/base.py
View file

@ -36,9 +36,7 @@ def get_langfuse_callback(trace_id):
return None
def flush_langfuse_callback_if_present(
callbacks: List[Union[BaseCallbackHandler, "CallbackHandler"]]
):
def flush_langfuse_callback_if_present(callbacks: List[Union[BaseCallbackHandler, "CallbackHandler"]]):
"""
If langfuse callback is present, run callback.langfuse.flush()
"""
@ -79,15 +77,9 @@ async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwa
# if langfuse callback is present, run callback.langfuse.flush()
flush_langfuse_callback_if_present(callbacks)
intermediate_steps = (
output.get("intermediate_steps", []) if isinstance(output, dict) else []
)
intermediate_steps = output.get("intermediate_steps", []) if isinstance(output, dict) else []
result = (
output.get(langchain_object.output_keys[0])
if isinstance(output, dict)
else output
)
result = output.get(langchain_object.output_keys[0]) if isinstance(output, dict) else output
try:
thought = format_actions(intermediate_steps) if intermediate_steps else ""
except Exception as exc:

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

@ -126,9 +126,7 @@ async def process_runnable(runnable: Runnable, inputs: Union[dict, List[dict]]):
elif isinstance(inputs, dict) and hasattr(runnable, "ainvoke"):
result = await runnable.ainvoke(inputs)
else:
raise ValueError(
f"Runnable {runnable} does not support inputs of type {type(inputs)}"
)
raise ValueError(f"Runnable {runnable} does not support inputs of type {type(inputs)}")
# Check if the result is a list of AIMessages
if isinstance(result, list) and all(isinstance(r, AIMessage) for r in result):
result = [r.content for r in result]
@ -137,9 +135,7 @@ async def process_runnable(runnable: Runnable, inputs: Union[dict, List[dict]]):
return result
async def process_inputs_dict(
built_object: Union[Chain, VectorStore, Runnable], inputs: dict
):
async def process_inputs_dict(built_object: Union[Chain, VectorStore, Runnable], inputs: dict):
if isinstance(built_object, Chain):
if inputs is None:
raise ValueError("Inputs must be provided for a Chain")
@ -174,9 +170,7 @@ async def process_inputs_list(built_object: Runnable, inputs: List[dict]):
return await process_runnable(built_object, inputs)
async def generate_result(
built_object: Union[Chain, VectorStore, Runnable], inputs: Union[dict, List[dict]]
):
async def generate_result(built_object: Union[Chain, VectorStore, Runnable], inputs: Union[dict, List[dict]]):
if isinstance(inputs, dict):
result = await process_inputs_dict(built_object, inputs)
elif isinstance(inputs, List) and isinstance(built_object, Runnable):
@ -214,9 +208,7 @@ async def run_graph(
else:
graph_data = graph._graph_data
if not session_id and session_service is not None:
session_id = session_service.generate_key(
session_id=flow_id, data_graph=graph_data
)
session_id = session_service.generate_key(session_id=flow_id, data_graph=graph_data)
if inputs is None:
inputs = {}
@ -226,18 +218,14 @@ async def run_graph(
return outputs, session_id
def validate_input(
graph_data: Dict[str, Any], tweaks: Dict[str, Dict[str, Any]]
) -> List[Dict[str, Any]]:
def validate_input(graph_data: Dict[str, Any], tweaks: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
if not isinstance(graph_data, dict) or not isinstance(tweaks, dict):
raise ValueError("graph_data and tweaks should be dictionaries")
nodes = graph_data.get("data", {}).get("nodes") or graph_data.get("nodes")
if not isinstance(nodes, list):
raise ValueError(
"graph_data should contain a list of nodes under 'data' key or directly under 'nodes' key"
)
raise ValueError("graph_data should contain a list of nodes under 'data' key or directly under 'nodes' key")
return nodes
@ -246,9 +234,7 @@ def apply_tweaks(node: Dict[str, Any], node_tweaks: Dict[str, Any]) -> None:
template_data = node.get("data", {}).get("node", {}).get("template")
if not isinstance(template_data, dict):
logger.warning(
f"Template data for node {node.get('id')} should be a dictionary"
)
logger.warning(f"Template data for node {node.get('id')} should be a dictionary")
return
for tweak_name, tweak_value in node_tweaks.items():
@ -263,9 +249,7 @@ def apply_tweaks_on_vertex(vertex: Vertex, node_tweaks: Dict[str, Any]) -> None:
vertex.params[tweak_name] = tweak_value
def process_tweaks(
graph_data: Dict[str, Any], tweaks: Dict[str, Dict[str, Any]]
) -> Dict[str, Any]:
def process_tweaks(graph_data: Dict[str, Any], tweaks: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
"""
This function is used to tweak the graph data using the node id and the tweaks dict.
@ -286,9 +270,7 @@ def process_tweaks(
if node_tweaks := tweaks.get(node_id):
apply_tweaks(node, node_tweaks)
else:
logger.warning(
"Each node should be a dictionary with an 'id' key of type str"
)
logger.warning("Each node should be a dictionary with an 'id' key of type str")
return graph_data
@ -300,8 +282,6 @@ def process_tweaks_on_graph(graph: Graph, tweaks: Dict[str, Dict[str, Any]]):
if node_tweaks := tweaks.get(node_id):
apply_tweaks_on_vertex(vertex, node_tweaks)
else:
logger.warning(
"Each node should be a Vertex with an 'id' attribute of type str"
)
logger.warning("Each node should be a Vertex with an 'id' attribute of type str")
return graph

8
src/backend/langflow/services/chat/utils.py
View file

@ -23,9 +23,7 @@ async def process_graph(
if build_result is None:
# Raise user facing error
raise ValueError(
"There was an error loading the langchain_object. Please, check all the nodes and try again."
)
raise ValueError("There was an error loading the langchain_object. Please, check all the nodes and try again.")
# Generate result and thought
try:
@ -52,7 +50,5 @@ async def process_graph(
raise e
async def run_build_result(
build_result: Any, chat_inputs: ChatMessage, client_id: str, session_id: str
):
async def run_build_result(build_result: Any, chat_inputs: ChatMessage, client_id: str, session_id: str):
return build_result(inputs=chat_inputs.message)

12
src/backend/langflow/services/monitor/schema.py
View file

@ -10,9 +10,7 @@ if TYPE_CHECKING:
class TransactionModel(BaseModel):
id: Optional[int] = Field(default=None, alias="id")
timestamp: Optional[datetime] = Field(
default_factory=datetime.now, alias="timestamp"
)
timestamp: Optional[datetime] = Field(default_factory=datetime.now, alias="timestamp")
source: str
target: str
target_args: dict
@ -53,12 +51,8 @@ class MessageModel(BaseModel):
@classmethod
def from_record(cls, record: "Record"):
# first check if the record has all the required fields
if not record.data or (
"sender" not in record.data and "sender_name" not in record.data
):
raise ValueError(
"The record does not have the required fields 'sender' and 'sender_name' in the data."
)
if not record.data or ("sender" not in record.data and "sender_name" not in record.data):
raise ValueError("The record does not have the required fields 'sender' and 'sender_name' in the data.")
return cls(
sender=record.data["sender"],
sender_name=record.data["sender_name"],

4
src/backend/langflow/services/monitor/service.py
View file

@ -44,9 +44,7 @@ class MonitorService(Service):
def ensure_tables_exist(self):
for table_name, model in self.table_map.items():
drop_and_create_table_if_schema_mismatch(
str(self.db_path), table_name, model
)
drop_and_create_table_if_schema_mismatch(str(self.db_path), table_name, model)
def add_row(
self,

17
src/backend/langflow/services/monitor/utils.py
View file

@ -23,10 +23,7 @@ def get_table_schema_as_dict(conn: duckdb.DuckDBPyConnection, table_name: str) -
def model_to_sql_column_definitions(model: Type[BaseModel]) -> dict:
columns = {}
for field_name, field_type in model.model_fields.items():
if (
hasattr(field_type.annotation, "__args__")
and field_type.annotation is not None
):
if hasattr(field_type.annotation, "__args__") and field_type.annotation is not None:
field_args = field_type.annotation.__args__
else:
field_args = []
@ -49,9 +46,7 @@ def model_to_sql_column_definitions(model: Type[BaseModel]) -> dict:
return columns
def drop_and_create_table_if_schema_mismatch(
db_path: str, table_name: str, model: Type[BaseModel]
):
def drop_and_create_table_if_schema_mismatch(db_path: str, table_name: str, model: Type[BaseModel]):
with duckdb.connect(db_path) as conn:
# Get the current schema from the database
try:
@ -72,12 +67,8 @@ def drop_and_create_table_if_schema_mismatch(
conn.execute(f"CREATE SEQUENCE seq_{table_name} START 1;")
except duckdb.CatalogException:
pass
desired_schema[INDEX_KEY] = (
f"INTEGER PRIMARY KEY DEFAULT NEXTVAL('seq_{table_name}')"
)
columns_sql = ", ".join(
f"{name} {data_type}" for name, data_type in desired_schema.items()
)
desired_schema[INDEX_KEY] = f"INTEGER PRIMARY KEY DEFAULT NEXTVAL('seq_{table_name}')"
columns_sql = ", ".join(f"{name} {data_type}" for name, data_type in desired_schema.items())
create_table_sql = f"CREATE TABLE {table_name} ({columns_sql})"
conn.execute(create_table_sql)

4
src/backend/langflow/services/settings/manager.py
View file

@ -32,9 +32,7 @@ class SettingsService(Service):
for key in settings_dict:
if key not in Settings.model_fields.keys():
raise KeyError(f"Key {key} not found in settings")
logger.debug(
f"Loading {len(settings_dict[key])} {key} from {file_path}"
)
logger.debug(f"Loading {len(settings_dict[key])} {key} from {file_path}")
settings = Settings(**settings_dict)
if not settings.CONFIG_DIR:

4
src/backend/langflow/services/socket/utils.py
View file

@ -96,9 +96,7 @@ async def build_vertex(
)
# Emit the vertex build response
response = VertexBuildResponse(
valid=valid, params=params, id=vertex.id, data=result_dict
)
response = VertexBuildResponse(valid=valid, params=params, id=vertex.id, data=result_dict)
await sio.emit("vertex_build", data=response.model_dump(), to=sid)
except Exception as exc:

22
src/backend/langflow/services/storage/s3.py
View file

@ -25,9 +25,7 @@ class S3StorageService(StorageService):
:raises Exception: If an error occurs during file saving.
"""
try:
self.s3_client.put_object(
Bucket=self.bucket, Key=f"{folder}/{file_name}", Body=data
)
self.s3_client.put_object(Bucket=self.bucket, Key=f"{folder}/{file_name}", Body=data)
logger.info(f"File {file_name} saved successfully in folder {folder}.")
except NoCredentialsError:
logger.error("Credentials not available for AWS S3.")
@ -46,12 +44,8 @@ class S3StorageService(StorageService):
:raises Exception: If an error occurs during file retrieval.
"""
try:
response = self.s3_client.get_object(
Bucket=self.bucket, Key=f"{folder}/{file_name}"
)
logger.info(
f"File {file_name} retrieved successfully from folder {folder}."
)
response = self.s3_client.get_object(Bucket=self.bucket, Key=f"{folder}/{file_name}")
logger.info(f"File {file_name} retrieved successfully from folder {folder}.")
return response["Body"].read()
except ClientError as e:
logger.error(f"Error retrieving file {file_name} from folder {folder}: {e}")
@ -67,11 +61,7 @@ class S3StorageService(StorageService):
"""
try:
response = self.s3_client.list_objects_v2(Bucket=self.bucket, Prefix=folder)
files = [
item["Key"]
for item in response.get("Contents", [])
if "/" not in item["Key"][len(folder) :]
]
files = [item["Key"] for item in response.get("Contents", []) if "/" not in item["Key"][len(folder) :]]
logger.info(f"{len(files)} files listed in folder {folder}.")
return files
except ClientError as e:
@ -87,9 +77,7 @@ class S3StorageService(StorageService):
:raises Exception: If an error occurs during file deletion.
"""
try:
self.s3_client.delete_object(
Bucket=self.bucket, Key=f"{folder}/{file_name}"
)
self.s3_client.delete_object(Bucket=self.bucket, Key=f"{folder}/{file_name}")
logger.info(f"File {file_name} deleted successfully from folder {folder}.")
except ClientError as e:
logger.error(f"Error deleting file {file_name} from folder {folder}: {e}")

4
src/backend/langflow/services/storage/service.py
View file

@ -11,9 +11,7 @@ if TYPE_CHECKING:
class StorageService(Service):
name = "storage_service"
def __init__(
self, session_service: "SessionService", settings_service: "SettingsService"
):
def __init__(self, session_service: "SessionService", settings_service: "SettingsService"):
self.settings_service = settings_service
self.session_service = session_service
self.set_ready()

10
src/backend/langflow/template/field/base.py
View file

@ -68,9 +68,7 @@ class TemplateField(BaseModel):
refresh: Optional[bool] = None
"""Specifies if the field should be refreshed. Defaults to False."""
range_spec: Optional[RangeSpec] = Field(
default=None, serialization_alias="rangeSpec"
)
range_spec: Optional[RangeSpec] = Field(default=None, serialization_alias="rangeSpec")
"""Range specification for the field. Defaults to None."""
title_case: bool = False
@ -119,10 +117,6 @@ class TemplateField(BaseModel):
if not isinstance(value, list):
raise ValueError("file_types must be a list")
return [
(
f".{file_type}"
if isinstance(file_type, str) and not file_type.startswith(".")
else file_type
)
(f".{file_type}" if isinstance(file_type, str) and not file_type.startswith(".") else file_type)
for file_type in value
]

27
src/backend/langflow/template/frontend_node/base.py
View file

@ -171,9 +171,7 @@ class FrontendNode(BaseModel):
return _type
@staticmethod
def handle_special_field(
field, key: str, _type: str, SPECIAL_FIELD_HANDLERS
) -> str:
def handle_special_field(field, key: str, _type: str, SPECIAL_FIELD_HANDLERS) -> str:
"""Handles special field by using the respective handler if present."""
handler = SPECIAL_FIELD_HANDLERS.get(key)
return handler(field) if handler else _type
@ -184,11 +182,7 @@ class FrontendNode(BaseModel):
if "dict" in _type.lower() and field.name == "dict_":
field.field_type = "file"
field.file_types = [".json", ".yaml", ".yml"]
elif (
_type.startswith("Dict")
or _type.startswith("Mapping")
or _type.startswith("dict")
):
elif _type.startswith("Dict") or _type.startswith("Mapping") or _type.startswith("dict"):
field.field_type = "dict"
return _type
@ -199,9 +193,7 @@ class FrontendNode(BaseModel):
field.value = value["default"]
@staticmethod
def handle_specific_field_values(
field: TemplateField, key: str, name: Optional[str] = None
) -> None:
def handle_specific_field_values(field: TemplateField, key: str, name: Optional[str] = None) -> None:
"""Handles specific field values for certain fields."""
if key == "headers":
field.value = """{"Authorization": "Bearer <token>"}"""
@ -209,9 +201,7 @@ class FrontendNode(BaseModel):
FrontendNode._handle_api_key_specific_field_values(field, key, name)
@staticmethod
def _handle_model_specific_field_values(
field: TemplateField, key: str, name: Optional[str] = None
) -> None:
def _handle_model_specific_field_values(field: TemplateField, key: str, name: Optional[str] = None) -> None:
"""Handles specific field values related to models."""
model_dict = {
"OpenAI": constants.OPENAI_MODELS,
@ -224,9 +214,7 @@ class FrontendNode(BaseModel):
field.is_list = True
@staticmethod
def _handle_api_key_specific_field_values(
field: TemplateField, key: str, name: Optional[str] = None
) -> None:
def _handle_api_key_specific_field_values(field: TemplateField, key: str, name: Optional[str] = None) -> None:
"""Handles specific field values related to API keys."""
if "api_key" in key and "OpenAI" in str(name):
field.display_name = "OpenAI API Key"
@ -266,10 +254,7 @@ class FrontendNode(BaseModel):
@staticmethod
def should_be_password(key: str, show: bool) -> bool:
"""Determines whether the field should be a password field."""
return (
any(text in key.lower() for text in {"password", "token", "api", "key"})
and show
)
return any(text in key.lower() for text in {"password", "token", "api", "key"}) and show
@staticmethod
def should_be_multiline(key: str) -> bool:

4
src/backend/langflow/template/frontend_node/chains.py
View file

@ -216,9 +216,7 @@ class MidJourneyPromptChainNode(FrontendNode):
),
],
)
description: str = (
"MidJourneyPromptChain is a chain you can use to generate new MidJourney prompts."
)
description: str = "MidJourneyPromptChain is a chain you can use to generate new MidJourney prompts."
base_classes: list[str] = [
"LLMChain",
"BaseCustomChain",

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

@ -25,10 +25,7 @@ def patching(record):
def configure(log_level: Optional[str] = None, log_file: Optional[Path] = None):
if (
os.getenv("LANGFLOW_LOG_LEVEL", "").upper() 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,9 +45,7 @@ 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:
@ -91,23 +89,15 @@ 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:
@ -134,23 +124,15 @@ 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]
@ -212,13 +194,9 @@ 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)
@ -239,11 +217,7 @@ 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
@ -256,9 +230,7 @@ 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
@ -302,9 +274,7 @@ 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

4
src/backend/langflow/worker.py
View file

@ -24,9 +24,7 @@ def build_vertex(self, vertex: "Vertex") -> "Vertex":
async_to_sync(vertex.build)()
return vertex
except SoftTimeLimitExceeded as e:
raise self.retry(
exc=SoftTimeLimitExceeded("Task took too long"), countdown=2
) from e
raise self.retry(exc=SoftTimeLimitExceeded("Task took too long"), countdown=2) from e
@celery_app.task(acks_late=True)

41
tests/test_endpoints.py
View file

@ -29,10 +29,7 @@ def poll_task_status(client, headers, href, max_attempts=20, sleep_time=1):
href,
headers=headers,
)
if (
task_status_response.status_code == 200
and task_status_response.json()["status"] == "SUCCESS"
):
if task_status_response.status_code == 200 and task_status_response.json()["status"] == "SUCCESS":
return task_status_response.json()
time.sleep(sleep_time)
return None # Return None if task did not complete in time
@ -126,11 +123,7 @@ def created_api_key(active_user):
)
db_manager = get_db_service()
with session_getter(db_manager) as session:
if (
existing_api_key := session.query(ApiKey)
.filter(ApiKey.api_key == api_key.api_key)
.first()
):
if existing_api_key := session.query(ApiKey).filter(ApiKey.api_key == api_key.api_key).first():
return existing_api_key
session.add(api_key)
session.commit()
@ -296,11 +289,7 @@ def test_get_all(client: TestClient, logged_in_headers):
dir_reader = DirectoryReader(settings.COMPONENTS_PATH[0])
files = dir_reader.get_files()
# json_response is a dict of dicts
all_names = [
component_name
for _, components in response.json().items()
for component_name in components
]
all_names = [component_name for _, components in response.json().items() for component_name in components]
json_response = response.json()
# We need to test the custom nodes
assert len(all_names) > len(files)
@ -425,19 +414,13 @@ def test_various_prompts(client, prompt, expected_input_variables):
def test_get_vertices_flow_not_found(client, logged_in_headers):
response = client.get(
"/api/v1/build/nonexistent_id/vertices", headers=logged_in_headers
)
assert (
response.status_code == 500
) # Or whatever status code you've set for invalid ID
response = client.get("/api/v1/build/nonexistent_id/vertices", headers=logged_in_headers)
assert response.status_code == 500 # Or whatever status code you've set for invalid ID
def test_get_vertices(client, added_flow_with_prompt_and_history, logged_in_headers):
flow_id = added_flow_with_prompt_and_history["id"]
response = client.get(
f"/api/v1/build/{flow_id}/vertices", headers=logged_in_headers
)
response = client.get(f"/api/v1/build/{flow_id}/vertices", headers=logged_in_headers)
assert response.status_code == 200
assert "ids" in response.json()
# The response should contain the list in this order
@ -453,19 +436,13 @@ def test_get_vertices(client, added_flow_with_prompt_and_history, logged_in_head
def test_build_vertex_invalid_flow_id(client, logged_in_headers):
response = client.post(
"/api/v1/build/nonexistent_id/vertices/vertex_id", headers=logged_in_headers
)
response = client.post("/api/v1/build/nonexistent_id/vertices/vertex_id", headers=logged_in_headers)
assert response.status_code == 500
def test_build_vertex_invalid_vertex_id(
client, added_flow_with_prompt_and_history, logged_in_headers
):
def test_build_vertex_invalid_vertex_id(client, added_flow_with_prompt_and_history, logged_in_headers):
flow_id = added_flow_with_prompt_and_history["id"]
response = client.post(
f"/api/v1/build/{flow_id}/vertices/invalid_vertex_id", headers=logged_in_headers
)
response = client.post(f"/api/v1/build/{flow_id}/vertices/invalid_vertex_id", headers=logged_in_headers)
assert response.status_code == 500