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

Merge branch 'dev' into ij/chatimg

Browse source
This commit is contained in:
Ítalo Johnny 2024-05-28 15:57:31 -03:00 • committed by GitHub
commit 67442774cf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
72 changed files with 5416 additions and 3559 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/backend/base/langflow/__main__.py
View file

@ -77,7 +77,7 @@ def set_var_for_macos_issue():
def run(
host: str = typer.Option("127.0.0.1", help="Host to bind the server to.", envvar="LANGFLOW_HOST"),
workers: int = typer.Option(1, help="Number of worker processes.", envvar="LANGFLOW_WORKERS"),
timeout: int = typer.Option(300, help="Worker timeout in seconds."),
timeout: int = typer.Option(300, help="Worker timeout in seconds.", envvar="LANGFLOW_WORKER_TIMEOUT"),
port: int = typer.Option(7860, help="Port to listen on.", envvar="LANGFLOW_PORT"),
components_path: Optional[Path] = typer.Option(
Path(__file__).parent / "components",
@ -145,6 +145,10 @@ def run(
if is_port_in_use(port, host):
port = get_free_port(port)
settings_service = get_settings_service()
settings_service.set("worker_timeout", timeout)
options = {
"bind": f"{host}:{port}",
"workers": get_number_of_workers(workers),

19
src/backend/base/langflow/api/v1/endpoints.py
View file

@ -1,5 +1,5 @@
from http import HTTPStatus
from typing import Annotated, List, Optional, Union
from typing import TYPE_CHECKING, Annotated, List, Optional, Union
from uuid import UUID
import sqlalchemy as sa
@ -9,6 +9,7 @@ from sqlmodel import Session, select
from langflow.api.utils import update_frontend_node_with_template_values
from langflow.api.v1.schemas import (
ConfigResponse,
CustomComponentRequest,
InputValueRequest,
ProcessResponse,
@ -31,7 +32,9 @@ from langflow.services.deps import get_session, get_session_service, get_setting
from langflow.services.session.service import SessionService
from langflow.services.task.service import TaskService
# build router
if TYPE_CHECKING:
from langflow.services.settings.manager import SettingsService
router = APIRouter(tags=["Base"])
@ -440,3 +443,15 @@ async def custom_component_update(
except Exception as exc:
logger.exception(exc)
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.get("/config", response_model=ConfigResponse)
def get_config():
try:
from langflow.services.deps import get_settings_service
settings_service: "SettingsService" = get_settings_service()
return settings_service.settings.model_dump()
except Exception as exc:
logger.exception(exc)
raise HTTPException(status_code=500, detail=str(exc)) from exc

4
src/backend/base/langflow/api/v1/schemas.py
View file

@ -322,3 +322,7 @@ class FlowDataRequest(BaseModel):
nodes: List[dict]
edges: List[dict]
viewport: Optional[dict] = None
class ConfigResponse(BaseModel):
frontend_timeout: int

0
src/backend/base/langflow/base/flow_processing/__init__.py Normal file
View file

67
src/backend/base/langflow/base/flow_processing/utils.py Normal file
View file

@ -0,0 +1,67 @@
from typing import List
from langflow.graph.schema import ResultData, RunOutputs
from langflow.schema.schema import Record
def build_records_from_run_outputs(run_outputs: RunOutputs) -> List[Record]:
"""
Build a list of records from the given RunOutputs.
Args:
run_outputs (RunOutputs): The RunOutputs object containing the output data.
Returns:
List[Record]: A list of records built from the RunOutputs.
"""
if not run_outputs:
return []
records = []
for result_data in run_outputs.outputs:
if result_data:
records.extend(build_records_from_result_data(result_data))
return records
def build_records_from_result_data(result_data: ResultData, get_final_results_only: bool = True) -> List[Record]:
"""
Build a list of records from the given ResultData.
Args:
result_data (ResultData): The ResultData object containing the result data.
get_final_results_only (bool, optional): Whether to include only final results. Defaults to True.
Returns:
List[Record]: A list of records built from the ResultData.
"""
messages = result_data.messages
if not messages:
return []
records = []
for message in messages:
message_dict = message if isinstance(message, dict) else message.model_dump()
if get_final_results_only:
result_data_dict = result_data.model_dump()
results = result_data_dict.get("results", {})
inner_result = results.get("result", {})
record = Record(data={"result": inner_result, "message": message_dict}, text_key="result")
records.append(record)
return records
def format_flow_output_records(records: List[Record]) -> str:
"""
Format the flow output records into a string.
Args:
records (List[Record]): The list of records to format.
Returns:
str: The formatted flow output records.
"""
result = "Flow run output:\n"
results = "\n".join([record.result for record in records if record.data["message"]])
return result + results

117
src/backend/base/langflow/base/tools/flow_tool.py Normal file
View file

@ -0,0 +1,117 @@
from typing import Any, List, Optional, Type
from asyncer import syncify
from langchain.tools import BaseTool
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import ToolException
from pydantic.v1 import BaseModel
from langflow.base.flow_processing.utils import build_records_from_result_data, format_flow_output_records
from langflow.graph.graph.base import Graph
from langflow.graph.vertex.base import Vertex
from langflow.helpers.flow import build_schema_from_inputs, get_arg_names, get_flow_inputs, run_flow
class FlowTool(BaseTool):
name: str
description: str
graph: Optional[Graph] = None
flow_id: Optional[str] = None
user_id: Optional[str] = None
inputs: List["Vertex"] = []
get_final_results_only: bool = True
@property
def args(self) -> dict:
schema = self.get_input_schema()
return schema.schema()["properties"]
def get_input_schema(self, config: Optional[RunnableConfig] = None) -> Type[BaseModel]:
"""The tool's input schema."""
if self.args_schema is not None:
return self.args_schema
elif self.graph is not None:
return build_schema_from_inputs(self.name, get_flow_inputs(self.graph))
else:
raise ToolException("No input schema available.")
def _run(
self,
*args: Any,
**kwargs: Any,
) -> str:
"""Use the tool."""
args_names = get_arg_names(self.inputs)
if len(args_names) == len(args):
kwargs = {arg["arg_name"]: arg_value for arg, arg_value in zip(args_names, args)}
elif len(args_names) != len(args) and len(args) != 0:
raise ToolException(
"Number of arguments does not match the number of inputs. Pass keyword arguments instead."
)
tweaks = {arg["component_name"]: kwargs[arg["arg_name"]] for arg in args_names}
run_outputs = syncify(run_flow, raise_sync_error=False)(
tweaks={key: {"input_value": value} for key, value in tweaks.items()},
flow_id=self.flow_id,
user_id=self.user_id,
)
if not run_outputs:
return "No output"
run_output = run_outputs[0]
records = []
if run_output is not None:
for output in run_output.outputs:
if output:
records.extend(
build_records_from_result_data(output, get_final_results_only=self.get_final_results_only)
)
return format_flow_output_records(records)
def validate_inputs(self, args_names: List[dict[str, str]], args: Any, kwargs: Any):
"""Validate the inputs."""
if len(args) > 0 and len(args) != len(args_names):
raise ToolException(
"Number of positional arguments does not match the number of inputs. Pass keyword arguments instead."
)
if len(args) == len(args_names):
kwargs = {arg_name["arg_name"]: arg_value for arg_name, arg_value in zip(args_names, args)}
missing_args = [arg["arg_name"] for arg in args_names if arg["arg_name"] not in kwargs]
if missing_args:
raise ToolException(f"Missing required arguments: {', '.join(missing_args)}")
return kwargs
def build_tweaks_dict(self, args, kwargs):
args_names = get_arg_names(self.inputs)
kwargs = self.validate_inputs(args_names=args_names, args=args, kwargs=kwargs)
tweaks = {arg["component_name"]: kwargs[arg["arg_name"]] for arg in args_names}
return tweaks
async def _arun(
self,
*args: Any,
**kwargs: Any,
) -> str:
"""Use the tool asynchronously."""
tweaks = self.build_tweaks_dict(args, kwargs)
run_outputs = await run_flow(
tweaks={key: {"input_value": value} for key, value in tweaks.items()},
flow_id=self.flow_id,
user_id=self.user_id,
)
if not run_outputs:
return "No output"
run_output = run_outputs[0]
records = []
if run_output is not None:
for output in run_output.outputs:
if output:
records.extend(
build_records_from_result_data(output, get_final_results_only=self.get_final_results_only)
)
return format_flow_output_records(records)

20
src/backend/base/langflow/components/experimental/FlowTool.py
View file

@ -1,14 +1,14 @@
from typing import Any, List, Optional
from asyncer import syncify
from langchain_core.tools import StructuredTool
from loguru import logger
from langflow.base.tools.flow_tool import FlowTool
from langflow.custom import CustomComponent
from langflow.field_typing import Tool
from langflow.graph.graph.base import Graph
from langflow.helpers.flow import build_function_and_schema
from langflow.helpers.flow import get_flow_inputs
from langflow.schema.dotdict import dotdict
from langflow.schema.schema import Record
from loguru import logger
class FlowToolComponent(CustomComponent):
@ -68,18 +68,20 @@ class FlowToolComponent(CustomComponent):
}
async def build(self, flow_name: str, name: str, description: str, return_direct: bool = False) -> Tool:
FlowTool.update_forward_refs()
flow_record = self.get_flow(flow_name)
if not flow_record:
raise ValueError("Flow not found.")
graph = Graph.from_payload(flow_record.data["data"])
dynamic_flow_function, schema = build_function_and_schema(flow_record, graph)
tool = StructuredTool.from_function(
func=syncify(dynamic_flow_function, raise_sync_error=False), # type: ignore
coroutine=dynamic_flow_function,
inputs = get_flow_inputs(graph)
tool = FlowTool(
name=name,
description=description,
graph=graph,
return_direct=return_direct,
args_schema=schema,
inputs=inputs,
flow_id=str(flow_record.id),
user_id=str(self._user_id),
)
description_repr = repr(tool.description).strip("'")
args_str = "\n".join([f"- {arg_name}: {arg_data['description']}" for arg_name, arg_data in tool.args.items()])

20
src/backend/base/langflow/components/experimental/RunFlow.py
View file

@ -1,8 +1,9 @@
from typing import Any, List, Optional
from langflow.base.flow_processing.utils import build_records_from_run_outputs
from langflow.custom import CustomComponent
from langflow.field_typing import NestedDict, Text
from langflow.graph.schema import ResultData
from langflow.graph.schema import RunOutputs
from langflow.schema import Record, dotdict
@ -39,28 +40,17 @@ class RunFlowComponent(CustomComponent):
},
}
def build_records_from_result_data(self, result_data: ResultData) -> List[Record]:
messages = result_data.messages
if not messages:
return []
records = []
for message in messages:
message_dict = message if isinstance(message, dict) else message.model_dump()
record = Record(text=message_dict.get("text", ""), data={"result": result_data})
records.append(record)
return records
async def build(self, input_value: Text, flow_name: str, tweaks: NestedDict) -> List[Record]:
results: List[Optional[ResultData]] = await self.run_flow(
results: List[Optional[RunOutputs]] = await self.run_flow(
inputs={"input_value": input_value}, flow_name=flow_name, tweaks=tweaks
)
if isinstance(results, list):
records = []
for result in results:
if result:
records.extend(self.build_records_from_result_data(result))
records.extend(build_records_from_run_outputs(result))
else:
records = self.build_records_from_result_data(results)
records = build_records_from_run_outputs()(results)
self.status = records
return records

20
src/backend/base/langflow/components/experimental/SubFlow.py
View file

@ -2,9 +2,10 @@ from typing import Any, List, Optional
from loguru import logger
from langflow.base.flow_processing.utils import build_records_from_result_data
from langflow.custom import CustomComponent
from langflow.graph.graph.base import Graph
from langflow.graph.schema import ResultData, RunOutputs
from langflow.graph.schema import RunOutputs
from langflow.graph.vertex.base import Vertex
from langflow.helpers.flow import get_flow_inputs
from langflow.schema import Record
@ -92,21 +93,6 @@ class SubFlowComponent(CustomComponent):
},
}
def build_records_from_result_data(self, result_data: ResultData, get_final_results_only: bool) -> List[Record]:
messages = result_data.messages
if not messages:
return []
records = []
for message in messages:
message_dict = message if isinstance(message, dict) else message.model_dump()
if get_final_results_only:
result_data_dict = result_data.model_dump()
results = result_data_dict.get("results", {})
inner_result = results.get("result", {})
record = Record(data={"result": inner_result, "message": message_dict}, text_key="result")
records.append(record)
return records
async def build(self, flow_name: str, get_final_results_only: bool = True, **kwargs) -> List[Record]:
tweaks = {key: {"input_value": value} for key, value in kwargs.items()}
run_outputs: List[Optional[RunOutputs]] = await self.run_flow(
@ -121,7 +107,7 @@ class SubFlowComponent(CustomComponent):
if run_output is not None:
for output in run_output.outputs:
if output:
records.extend(self.build_records_from_result_data(output, get_final_results_only))
records.extend(build_records_from_result_data(output, get_final_results_only))
self.status = records
logger.debug(records)

188
src/backend/base/langflow/components/models/OllamaModel.py
View file

@ -1,16 +1,21 @@
from typing import Dict, List, Optional
# from langchain_community.chat_models import ChatOllama
from langchain_community.chat_models import ChatOllama
from typing import Any, Dict, List, Optional, Union
from langchain_community.chat_models.ollama import ChatOllama
from langflow.base.constants import STREAM_INFO_TEXT
from langflow.base.models.model import LCModelComponent
from langchain_core.caches import BaseCache
# from langchain.chat_models import ChatOllama
from langflow.field_typing import Text
# whe When a callback component is added to Langflow, the comment must be uncommented.
# from langchain.callbacks.manager import CallbackManager
import asyncio
import json
import httpx
class ChatOllamaComponent(LCModelComponent):
@ -20,11 +25,19 @@ class ChatOllamaComponent(LCModelComponent):
field_order = [
"base_url",
"headers",
"keep_alive_flag",
"keep_alive",
"metadata",
"model",
"temperature",
"cache",
"callback_manager",
"callbacks",
"format",
"metadata",
"mirostat",
@ -54,12 +67,41 @@ class ChatOllamaComponent(LCModelComponent):
"base_url": {
"display_name": "Base URL",
"info": "Endpoint of the Ollama API. Defaults to 'http://localhost:11434' if not specified.",
},
"format": {
"display_name": "Format",
"info": "Specify the format of the output (e.g., json)",
"advanced": True,
},
"headers": {
"display_name": "Headers",
"advanced": True,
},
"keep_alive_flag": {
"display_name": "Unload interval",
"options": ["Keep", "Immediately","Minute", "Hour", "sec" ],
"real_time_refresh": True,
"refresh_button": True,
},
"keep_alive": {
"display_name": "interval",
"info": "How long the model will stay loaded into memory.",
},
"model": {
"display_name": "Model Name",
"value": "llama2",
"options": [],
"info": "Refer to https://ollama.ai/library for more models.",
"real_time_refresh": True,
"refresh_button": True,
},
"temperature": {
"display_name": "Temperature",
@ -67,25 +109,8 @@ class ChatOllamaComponent(LCModelComponent):
"value": 0.8,
"info": "Controls the creativity of model responses.",
},
"cache": {
"display_name": "Cache",
"field_type": "bool",
"info": "Enable or disable caching.",
"advanced": True,
"value": False,
},
### When a callback component is added to Langflow, the comment must be uncommented. ###
# "callback_manager": {
# "display_name": "Callback Manager",
# "info": "Optional callback manager for additional functionality.",
# "advanced": True,
# },
# "callbacks": {
# "display_name": "Callbacks",
# "info": "Callbacks to execute during model runtime.",
# "advanced": True,
# },
########################################################################################
"format": {
"display_name": "Format",
"field_type": "str",
@ -101,20 +126,24 @@ class ChatOllamaComponent(LCModelComponent):
"display_name": "Mirostat",
"options": ["Disabled", "Mirostat", "Mirostat 2.0"],
"info": "Enable/disable Mirostat sampling for controlling perplexity.",
"value": "Disabled",
"advanced": True,
"advanced": False,
"real_time_refresh": True,
"refresh_button": True,
},
"mirostat_eta": {
"display_name": "Mirostat Eta",
"field_type": "float",
"info": "Learning rate for Mirostat algorithm. (Default: 0.1)",
"advanced": True,
"real_time_refresh": True,
},
"mirostat_tau": {
"display_name": "Mirostat Tau",
"field_type": "float",
"info": "Controls the balance between coherence and diversity of the output. (Default: 5.0)",
"advanced": True,
"real_time_refresh": True,
},
"num_ctx": {
"display_name": "Context Window Size",
@ -211,21 +240,74 @@ class ChatOllamaComponent(LCModelComponent):
},
}
def update_build_config(self, build_config: dict, field_value: Any, field_name: str | None = None):
if field_name == "mirostat":
if field_value == "Disabled":
build_config["mirostat_eta"]["advanced"] = True
build_config["mirostat_tau"]["advanced"] = True
build_config["mirostat_eta"]["value"] = None
build_config["mirostat_tau"]["value"] = None
else:
build_config["mirostat_eta"]["advanced"] = False
build_config["mirostat_tau"]["advanced"] = False
if field_value == "Mirostat 2.0":
build_config["mirostat_eta"]["value"] = 0.2
build_config["mirostat_tau"]["value"] = 10
else:
build_config["mirostat_eta"]["value"] = 0.1
build_config["mirostat_tau"]["value"] = 5
if field_name == "model":
base_url = build_config.get("base_url", {}).get(
"value", "http://localhost:11434")
build_config["model"]["options"] = self.get_model(
base_url + "/api/tags")
if field_name == "keep_alive_flag":
if field_value == "Keep":
build_config["keep_alive"]["value"] = "-1"
build_config["keep_alive"]["advanced"] = True
elif field_value == "Immediately":
build_config["keep_alive"]["value"] = "0"
build_config["keep_alive"]["advanced"] = True
else:
build_config["keep_alive"]["advanced"] = False
return build_config
def get_model(self, url: str) -> List[str]:
try:
with httpx.Client() as client:
response = client.get(url)
response.raise_for_status()
data = response.json()
model_names = [model['name']
for model in data.get("models", [])]
return model_names
except Exception as e:
raise ValueError("Could not retrieve models") from e
return [""]
def build(
self,
base_url: Optional[str],
model: str,
input_value: Text,
mirostat: Optional[str],
mirostat: Optional[str],
mirostat_eta: Optional[float] = None,
mirostat_tau: Optional[float] = None,
### When a callback component is added to Langflow, the comment must be uncommented.###
# callback_manager: Optional[CallbackManager] = None,
# callbacks: Optional[List[Callbacks]] = None,
#######################################################################################
repeat_last_n: Optional[int] = None,
verbose: Optional[bool] = None,
cache: Optional[bool] = None,
keep_alive: Optional[int] = None,
keep_alive_flag: Optional[str] = None,
num_ctx: Optional[int] = None,
num_gpu: Optional[int] = None,
format: Optional[str] = None,
@ -244,33 +326,39 @@ class ChatOllamaComponent(LCModelComponent):
stream: bool = False,
system_message: Optional[str] = None,
) -> Text:
if not base_url:
base_url = "http://localhost:11434"
# Mapping mirostat settings to their corresponding values
mirostat_options = {"Mirostat": 1, "Mirostat 2.0": 2}
# Default to 0 for 'Disabled'
mirostat_value = mirostat_options.get(mirostat, 0) # type: ignore
# Set mirostat_eta and mirostat_tau to None if mirostat is disabled
if mirostat_value == 0:
mirostat_eta = None
mirostat_tau = None
if keep_alive_flag == "Minute":
keep_alive_instance = f"{keep_alive}m"
elif keep_alive_flag == "Hour":
keep_alive_instance = f"{keep_alive}h"
elif keep_alive_flag == "sec":
keep_alive_instance = f"{keep_alive}s"
elif keep_alive_flag == "Keep":
keep_alive_instance = "-1"
elif keep_alive_flag == "Immediately":
keep_alive_instance = "0"
else:
keep_alive_instance = "Invalid option"
mirostat_instance = 0
if mirostat == "disable":
mirostat_instance = 0
# Mapping system settings to their corresponding values
llm_params = {
"base_url": base_url,
"cache": cache,
"model": model,
"mirostat": mirostat_value,
"mirostat": mirostat_instance,
"keep_alive": keep_alive_instance,
"format": format,
"metadata": metadata,
"tags": tags,
## When a callback component is added to Langflow, the comment must be uncommented.##
# "callback_manager": callback_manager,
# "callbacks": callbacks,
#####################################################################################
"mirostat_eta": mirostat_eta,
"mirostat_tau": mirostat_tau,
"num_ctx": num_ctx,

6
src/backend/base/langflow/components/vectorsearch/ChromaSearch.py
View file

@ -1,5 +1,6 @@
from typing import List, Optional
import chromadb
from chromadb.config import Settings
from langchain_chroma import Chroma
@ -91,7 +92,7 @@ class ChromaSearchComponent(LCVectorStoreComponent):
# Chroma settings
chroma_settings = None
client = None
if chroma_server_host is not None:
chroma_settings = Settings(
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins or [],
@ -100,13 +101,14 @@ class ChromaSearchComponent(LCVectorStoreComponent):
chroma_server_grpc_port=chroma_server_grpc_port or None,
chroma_server_ssl_enabled=chroma_server_ssl_enabled,
)
client = chromadb.HttpClient(settings=chroma_settings)
if index_directory:
index_directory = self.resolve_path(index_directory)
vector_store = Chroma(
embedding_function=embedding,
collection_name=collection_name,
persist_directory=index_directory,
client_settings=chroma_settings,
client=client,
)
return self.search_with_vector_store(input_value, search_type, vector_store, k=number_of_results)

8
src/backend/base/langflow/components/vectorstores/Chroma.py
View file

@ -1,5 +1,6 @@
from typing import List, Optional, Union
import chromadb
from chromadb.config import Settings
from langchain_chroma import Chroma
from langchain_core.embeddings import Embeddings
@ -81,7 +82,7 @@ class ChromaComponent(CustomComponent):
# Chroma settings
chroma_settings = None
client = None
if chroma_server_host is not None:
chroma_settings = Settings(
chroma_server_cors_allow_origins=chroma_server_cors_allow_origins or [],
@ -90,6 +91,7 @@ class ChromaComponent(CustomComponent):
chroma_server_grpc_port=chroma_server_grpc_port or None,
chroma_server_ssl_enabled=chroma_server_ssl_enabled,
)
client = chromadb.HttpClient(settings=chroma_settings)
# If documents, then we need to create a Chroma instance using .from_documents
@ -111,12 +113,12 @@ class ChromaComponent(CustomComponent):
persist_directory=index_directory,
collection_name=collection_name,
embedding=embedding,
client_settings=chroma_settings,
client=client,
)
else:
chroma = Chroma(
persist_directory=index_directory,
client_settings=chroma_settings,
client=client,
embedding_function=embedding,
)
return chroma

53
src/backend/base/langflow/helpers/flow.py
View file

@ -1,10 +1,13 @@
from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Optional, Tuple, Type, Union, cast
from uuid import UUID
from pydantic.v1 import BaseModel, Field, create_model
from sqlmodel import select
from langflow.graph.schema import RunOutputs
from langflow.schema.schema import INPUT_FIELD_NAME, Record
from langflow.services.database.models.flow.model import Flow
from langflow.services.deps import session_scope
from pydantic.v1 import BaseModel, Field, create_model
from sqlmodel import select
if TYPE_CHECKING:
from langflow.graph.graph.base import Graph
@ -51,7 +54,7 @@ async def load_flow(
raise ValueError(f"Flow {flow_id} not found")
if tweaks:
graph_data = process_tweaks(graph_data=graph_data, tweaks=tweaks)
graph = Graph.from_payload(graph_data, flow_id=flow_id)
graph = Graph.from_payload(graph_data, flow_id=flow_id, user_id=user_id)
return graph
@ -67,25 +70,29 @@ async def run_flow(
flow_id: Optional[str] = None,
flow_name: Optional[str] = None,
user_id: Optional[str] = None,
) -> Any:
) -> List[RunOutputs]:
if user_id is None:
raise ValueError("Session is invalid")
graph = await load_flow(user_id, flow_id, flow_name, tweaks)
if inputs is None:
inputs = []
if isinstance(inputs, dict):
inputs = [inputs]
inputs_list = []
inputs_components = []
types = []
for input_dict in inputs:
inputs_list.append({INPUT_FIELD_NAME: cast(str, input_dict.get("input_value"))})
inputs_components.append(input_dict.get("components", []))
types.append(input_dict.get("type", []))
types.append(input_dict.get("type", "chat"))
return await graph.arun(inputs_list, inputs_components=inputs_components, types=types)
def generate_function_for_flow(inputs: List["Vertex"], flow_id: str) -> Callable[..., Awaitable[Any]]:
def generate_function_for_flow(
inputs: List["Vertex"], flow_id: str, user_id: str | UUID | None
) -> Callable[..., Awaitable[Any]]:
"""
Generate a dynamic flow function based on the given inputs and flow ID.
@ -129,11 +136,23 @@ async def flow_function({func_args}):
tweaks = {{ {arg_mappings} }}
from langflow.helpers.flow import run_flow
from langchain_core.tools import ToolException
from langflow.base.flow_processing.utils import build_records_from_result_data, format_flow_output_records
try:
return await run_flow(
run_outputs = await run_flow(
tweaks={{key: {{'input_value': value}} for key, value in tweaks.items()}},
flow_id="{flow_id}",
user_id="{user_id}"
)
if not run_outputs:
return []
run_output = run_outputs[0]
records = []
if run_output is not None:
for output in run_output.outputs:
if output:
records.extend(build_records_from_result_data(output, get_final_results_only=True))
return format_flow_output_records(records)
except Exception as e:
raise ToolException(f'Error running flow: ' + e)
"""
@ -145,7 +164,7 @@ async def flow_function({func_args}):
def build_function_and_schema(
flow_record: Record, graph: "Graph"
flow_record: Record, graph: "Graph", user_id: str | UUID | None
) -> Tuple[Callable[..., Awaitable[Any]], Type[BaseModel]]:
"""
Builds a dynamic function and schema for a given flow.
@ -159,7 +178,7 @@ def build_function_and_schema(
"""
flow_id = flow_record.id
inputs = get_flow_inputs(graph)
dynamic_flow_function = generate_function_for_flow(inputs, flow_id)
dynamic_flow_function = generate_function_for_flow(inputs, flow_id, user_id=user_id)
schema = build_schema_from_inputs(flow_record.name, inputs)
return dynamic_flow_function, schema
@ -200,3 +219,19 @@ def build_schema_from_inputs(name: str, inputs: List["Vertex"]) -> Type[BaseMode
description = input_.description
fields[field_name] = (str, Field(default="", description=description))
return create_model(name, **fields) # type: ignore
def get_arg_names(inputs: List["Vertex"]) -> List[dict[str, str]]:
"""
Returns a list of dictionaries containing the component name and its corresponding argument name.
Args:
inputs (List[Vertex]): A list of Vertex objects representing the inputs.
Returns:
List[dict[str, str]]: A list of dictionaries, where each dictionary contains the component name and its argument name.
"""
return [
{"component_name": input_.display_name, "arg_name": input_.display_name.lower().replace(" ", "_")}
for input_ in inputs
]

6
src/backend/base/langflow/initial_setup/starter_projects/VectorStore-RAG-Flows.json
View file

@ -572,7 +572,7 @@
"advanced": false,
"dynamic": false,
"info": "",
"load_from_db": false,
"load_from_db": true,
"title_case": false,
"input_types": [
"Text"
@ -965,7 +965,7 @@
"advanced": false,
"dynamic": false,
"info": "The OpenAI API Key to use for the OpenAI model.",
"load_from_db": false,
"load_from_db": true,
"title_case": false,
"input_types": [
"Text"
@ -2846,7 +2846,7 @@
"advanced": false,
"dynamic": false,
"info": "",
"load_from_db": false,
"load_from_db": true,
"title_case": false,
"input_types": [
"Text"

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

@ -104,6 +104,10 @@ class Settings(BaseSettings):
"""Whether to store environment variables as Global Variables in the database."""
variables_to_get_from_environment: list[str] = VARIABLES_TO_GET_FROM_ENVIRONMENT
"""List of environment variables to get from the environment and store in the database."""
worker_timeout: int = 300
"""Timeout for the API calls in seconds."""
frontend_timeout: int = 0
"""Timeout for the frontend API calls in seconds."""
@field_validator("config_dir", mode="before")
def set_langflow_dir(cls, value):

4
src/backend/base/langflow/services/settings/service.py
View file

@ -42,3 +42,7 @@ class SettingsService(Service):
CONFIG_DIR=settings.config_dir,
)
return cls(settings, auth_settings)
def set(self, key, value):
setattr(self.settings, key, value)
return self.settings

2
src/backend/base/pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "langflow-base"
version = "0.0.49"
version = "0.0.50"
description = "A Python package with a built-in web application"
authors = ["Langflow <contact@langflow.org>"]
maintainers = [

10
src/frontend/src/App.tsx
View file

@ -1,3 +1,4 @@
import axios from "axios";
import { useContext, useEffect, useState } from "react";
import { ErrorBoundary } from "react-error-boundary";
import { useNavigate } from "react-router-dom";
@ -15,6 +16,7 @@ import {
} from "./constants/constants";
import { AuthContext } from "./contexts/authContext";
import { autoLogin, getGlobalVariables, getHealth } from "./controllers/API";
import { setupAxiosDefaults } from "./controllers/API/utils";
import useTrackLastVisitedPath from "./hooks/use-track-last-visited-path";
import Router from "./routes";
import useAlertStore from "./stores/alertStore";
@ -28,10 +30,10 @@ export default function App() {
useTrackLastVisitedPath();
const removeFromTempNotificationList = useAlertStore(
(state) => state.removeFromTempNotificationList,
(state) => state.removeFromTempNotificationList
);
const tempNotificationList = useAlertStore(
(state) => state.tempNotificationList,
(state) => state.tempNotificationList
);
const [fetchError, setFetchError] = useState(false);
const isLoading = useFlowsManagerStore((state) => state.isLoading);
@ -49,7 +51,7 @@ export default function App() {
const refreshVersion = useDarkStore((state) => state.refreshVersion);
const refreshStars = useDarkStore((state) => state.refreshStars);
const setGlobalVariables = useGlobalVariablesStore(
(state) => state.setGlobalVariables,
(state) => state.setGlobalVariables
);
const checkHasStore = useStoreStore((state) => state.checkHasStore);
const navigate = useNavigate();
@ -114,9 +116,11 @@ export default function App() {
return new Promise<void>(async (resolve, reject) => {
if (isAuthenticated) {
try {
await setupAxiosDefaults();
await getFoldersApi();
await getTypes();
await refreshFlows();
console.log(axios.defaults);
const res = await getGlobalVariables();
setGlobalVariables(res);
checkHasStore();

10
src/frontend/src/controllers/API/api.tsx
View file

@ -48,7 +48,7 @@ function ApiInterceptor() {
}
await clearBuildVerticesState(error);
return Promise.reject(error);
},
}
);
const isAuthorizedURL = (url) => {
@ -65,10 +65,10 @@ function ApiInterceptor() {
const parsedURL = new URL(url);
const isDomainAllowed = authorizedDomains.some(
(domain) => parsedURL.origin === new URL(domain).origin,
(domain) => parsedURL.origin === new URL(domain).origin
);
const isEndpointAllowed = authorizedEndpoints.some((endpoint) =>
parsedURL.pathname.includes(endpoint),
parsedURL.pathname.includes(endpoint)
);
return isDomainAllowed || isEndpointAllowed;
@ -112,7 +112,7 @@ function ApiInterceptor() {
},
(error) => {
return Promise.reject(error);
},
}
);
return () => {
@ -144,7 +144,7 @@ function ApiInterceptor() {
if (error?.config?.headers) {
delete error.config.headers["Authorization"];
error.config.headers["Authorization"] = `Bearer ${cookies.get(
"access_token_lf",
"access_token_lf"
)}`;
const response = await axios.request(error.config);
return response;

32
src/frontend/src/controllers/API/utils.tsx Normal file
View file

@ -0,0 +1,32 @@
import axios from "axios";
import { BASE_URL_API } from "../../constants/constants";
/**
* Fetches the configuration data from the API.
* @returns {Promise<any>} A promise that resolves to the configuration data.
* @throws {Error} If there was an error fetching the configuration data.
*/
export async function fetchConfig() {
try {
const response = await axios.get(`${BASE_URL_API}config`);
return response.data;
} catch (error) {
console.error("Failed to fetch configuration:", error);
throw error;
}
}
/**
* Sets up default configurations for Axios.
* Fetches the timeout configuration and sets it as the default timeout for Axios requests.
*/
export async function setupAxiosDefaults() {
const config = await fetchConfig();
// Create Axios instance with the fetched timeout configuration
const timeoutInMilliseconds = config.frontend_timeout
? config.frontend_timeout * 1000
: 30000;
axios.defaults.baseURL = "";
axios.defaults.timeout = timeoutInMilliseconds;
}

74
src/frontend/src/modals/apiModal/utils/get-python-api-code.tsx
View file

@ -8,51 +8,51 @@
export default function getPythonApiCode(
flowId: string,
isAuth: boolean,
tweaksBuildedObject,
tweaksBuildedObject
): string {
const tweaksObject = tweaksBuildedObject[0];
return `import requests
from typing import Optional
from typing import Optional
BASE_API_URL = "${window.location.protocol}//${window.location.host}/api/v1/run"
FLOW_ID = "${flowId}"
# You can tweak the flow by adding a tweaks dictionary
# e.g {"OpenAI-XXXXX": {"model_name": "gpt-4"}}
TWEAKS = ${JSON.stringify(tweaksObject, null, 2)}
BASE_API_URL = "${window.location.protocol}//${window.location.host}/api/v1/run"
FLOW_ID = "${flowId}"
# You can tweak the flow by adding a tweaks dictionary
# e.g {"OpenAI-XXXXX": {"model_name": "gpt-4"}}
TWEAKS = ${JSON.stringify(tweaksObject, null, 2)}
def run_flow(message: str,
flow_id: str,
output_type: str = "chat",
input_type: str = "chat",
tweaks: Optional[dict] = None,
api_key: Optional[str] = None) -> dict:
"""
Run a flow with a given message and optional tweaks.
def run_flow(message: str,
flow_id: str,
output_type: str = "chat",
input_type: str = "chat",
tweaks: Optional[dict] = None,
api_key: Optional[str] = None) -> dict:
"""
Run a flow with a given message and optional tweaks.
:param message: The message to send to the flow
:param flow_id: The ID of the flow to run
:param tweaks: Optional tweaks to customize the flow
:return: The JSON response from the flow
"""
api_url = f"{BASE_API_URL}/{flow_id}"
:param message: The message to send to the flow
:param flow_id: The ID of the flow to run
:param tweaks: Optional tweaks to customize the flow
:return: The JSON response from the flow
"""
api_url = f"{BASE_API_URL}/{flow_id}"
payload = {
"input_value": message,
"output_type": output_type,
"input_type": input_type,
}
headers = None
if tweaks:
payload["tweaks"] = tweaks
if api_key:
headers = {"x-api-key": api_key}
response = requests.post(api_url, json=payload, headers=headers)
return response.json()
payload = {
"input_value": message,
"output_type": output_type,
"input_type": input_type,
}
headers = None
if tweaks:
payload["tweaks"] = tweaks
if api_key:
headers = {"x-api-key": api_key}
response = requests.post(api_url, json=payload, headers=headers)
return response.json()
# Setup any tweaks you want to apply to the flow
message = "message"
${!isAuth ? `api_key = "<your api key>"` : ""}
print(run_flow(message=message, flow_id=FLOW_ID, tweaks=TWEAKS${
# Setup any tweaks you want to apply to the flow
message = "message"
${!isAuth ? `api_key = "<your api key>"` : ""}
print(run_flow(message=message, flow_id=FLOW_ID, tweaks=TWEAKS${
!isAuth ? `, api_key=api_key` : ""
}))`;
}

12
src/frontend/src/modals/apiModal/utils/get-python-code.tsx
View file

@ -6,15 +6,15 @@
*/
export default function getPythonCode(
flowName: string,
tweaksBuildedObject,
tweaksBuildedObject
): string {
const tweaksObject = tweaksBuildedObject[0];
return `from langflow.load import run_flow_from_json
TWEAKS = ${JSON.stringify(tweaksObject, null, 2)}
TWEAKS = ${JSON.stringify(tweaksObject, null, 2)}
result = run_flow_from_json(flow="${flowName}.json",
input_value="message",
fallback_to_env_vars=True, # False by default
tweaks=TWEAKS)`;
result = run_flow_from_json(flow="${flowName}.json",
input_value="message",
fallback_to_env_vars=True, # False by default
tweaks=TWEAKS)`;
}

26
src/frontend/src/pages/FlowPage/components/nodeToolbarComponent/index.tsx
View file

@ -271,7 +271,7 @@ export default function NodeToolbarComponent({
selected &&
(hasApiKey || hasStore) &&
(event.ctrlKey || event.metaKey) &&
event.key === "u"
event.key.toUpperCase() === "U"
) {
event.preventDefault();
handleSelectChange("update");
@ -280,7 +280,7 @@ export default function NodeToolbarComponent({
selected &&
isGroup &&
(event.ctrlKey || event.metaKey) &&
event.key === "g"
event.key.toUpperCase() === "G"
) {
event.preventDefault();
handleSelectChange("ungroup");
@ -290,7 +290,7 @@ export default function NodeToolbarComponent({
(hasApiKey || hasStore) &&
(event.ctrlKey || event.metaKey) &&
event.shiftKey &&
event.key === "S"
event.key.toUpperCase() === "S"
) {
event.preventDefault();
setShowconfirmShare((state) => !state);
@ -300,7 +300,7 @@ export default function NodeToolbarComponent({
selected &&
(event.ctrlKey || event.metaKey) &&
event.shiftKey &&
event.key === "Q"
event.key.toUpperCase() === "Q"
) {
event.preventDefault();
if (isMinimal) {
@ -317,7 +317,7 @@ export default function NodeToolbarComponent({
selected &&
(event.ctrlKey || event.metaKey) &&
event.shiftKey &&
event.key === "U"
event.key.toUpperCase() === "U"
) {
event.preventDefault();
if (hasCode) return setOpenModal((state) => !state);
@ -327,12 +327,16 @@ export default function NodeToolbarComponent({
selected &&
(event.ctrlKey || event.metaKey) &&
event.shiftKey &&
event.key === "A"
event.key.toUpperCase() === "A"
) {
event.preventDefault();
setShowModalAdvanced((state) => !state);
}
if (selected && (event.ctrlKey || event.metaKey) && event.key === "s") {
if (
selected &&
(event.ctrlKey || event.metaKey) &&
event.key.toUpperCase() === "S"
) {
if (isSaved) {
event.preventDefault();
return setShowOverrideModal((state) => !state);
@ -347,7 +351,7 @@ export default function NodeToolbarComponent({
selected &&
(event.ctrlKey || event.metaKey) &&
event.shiftKey &&
event.key === "D"
event.key.toUpperCase() === "D"
) {
event.preventDefault();
if (data.node?.documentation) {
@ -357,7 +361,11 @@ export default function NodeToolbarComponent({
title: `${data.id} docs is not available at the moment.`,
});
}
if (selected && (event.ctrlKey || event.metaKey) && event.key === "j") {
if (
selected &&
(event.ctrlKey || event.metaKey) &&
event.key.toUpperCase() === "J"
) {
event.preventDefault();
downloadNode(flowComponent!);
}