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

merge fix

Browse source
This commit is contained in:
cristhianzl 2024-05-29 20:15:13 -03:00
commit e6f466188c
75 changed files with 3041 additions and 518 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

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

@ -246,7 +246,7 @@ class VerticesOrderResponse(BaseModel):
class Log(TypedDict):
message: str
message: Union[dict, str]
type: str
@ -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

27
src/backend/base/langflow/base/models/model.py
View file

@ -53,19 +53,28 @@ class LCModelComponent(CustomComponent):
key in response_metadata["token_usage"] for key in inner_openai_keys
):
token_usage = response_metadata["token_usage"]
completion_tokens = token_usage["completion_tokens"]
prompt_tokens = token_usage["prompt_tokens"]
total_tokens = token_usage["total_tokens"]
finish_reason = response_metadata["finish_reason"]
status_message = f"Tokens:\nInput: {prompt_tokens}\nOutput: {completion_tokens}\nTotal Tokens: {total_tokens}\nStop Reason: {finish_reason}\nResponse: {content}"
status_message = {
"tokens": {
"input": token_usage["prompt_tokens"],
"output": token_usage["completion_tokens"],
"total": token_usage["total_tokens"],
"stop_reason": response_metadata["finish_reason"],
"response": content,
}
}
elif all(key in response_metadata for key in anthropic_keys) and all(
key in response_metadata["usage"] for key in inner_anthropic_keys
):
usage = response_metadata["usage"]
input_tokens = usage["input_tokens"]
output_tokens = usage["output_tokens"]
stop_reason = response_metadata["stop_reason"]
status_message = f"Tokens:\nInput: {input_tokens}\nOutput: {output_tokens}\nStop Reason: {stop_reason}\nResponse: {content}"
status_message = {
"tokens": {
"input": usage["input_tokens"],
"output": usage["output_tokens"],
"stop_reason": response_metadata["stop_reason"],
"response": content,
}
}
else:
status_message = f"Response: {content}"
else:

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

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

@ -734,7 +734,7 @@ class Graph:
await vertex.build(user_id=user_id, inputs=inputs_dict,files=files, fallback_to_env_vars=fallback_to_env_vars)
if vertex.result is not None:
params = vertex._built_object_repr()
params = vertex.artifacts_raw
log_type = vertex.artifacts_type
valid = True
result_dict = vertex.result

11
src/backend/base/langflow/graph/utils.py
View file

@ -19,6 +19,7 @@ class UnbuiltResult:
class ArtifactType(str, Enum):
TEXT = "text"
RECORD = "record"
OBJECT = "object"
UNKNOWN = "unknown"
@ -60,16 +61,16 @@ def serialize_field(value):
return value
def get_artifact_type(build_result: Any) -> str:
result = None
match build_result:
def get_artifact_type(value: Any) -> str:
result = ArtifactType.UNKNOWN
match value:
case Record():
result = ArtifactType.RECORD
case str():
result = ArtifactType.TEXT
case _:
result = ArtifactType.UNKNOWN
case dict():
result = ArtifactType.OBJECT
return result.value

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

@ -63,6 +63,7 @@ class Vertex:
self._built_result = None
self._built = False
self.artifacts: Dict[str, Any] = {}
self.artifacts_raw: Any = None
self.artifacts_type: Optional[str] = None
self.steps: List[Callable] = [self._build]
self.steps_ran: List[Callable] = []
@ -626,6 +627,7 @@ class Vertex:
self._built_object, self.artifacts = result
elif len(result) == 3:
self._custom_component, self._built_object, self.artifacts = result
self.artifacts_raw = self.artifacts.get("raw")
self.artifacts_type = self.artifacts.get("type") or ArtifactType.UNKNOWN.value
else:

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
]

10
src/backend/base/langflow/interface/initialize/loading.py
View file

@ -126,5 +126,13 @@ async def instantiate_custom_component(params, user_id, vertex, fallback_to_env_
custom_repr = build_result
if not isinstance(custom_repr, str):
custom_repr = str(custom_repr)
artifact = {"repr": custom_repr, "type": get_artifact_type(build_result)}
raw = custom_component.repr_value
if hasattr(raw, "data"):
raw = raw.data
elif hasattr(raw, "model_dump"):
raw = raw.model_dump()
artifact = {"repr": custom_repr, "raw": raw, "type": get_artifact_type(custom_component.repr_value)}
return custom_component, build_result, artifact

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 = [