diff --git a/src/backend/base/langflow/api/v1/validate.py b/src/backend/base/langflow/api/v1/validate.py index 97841bca6..2954fd13a 100644 --- a/src/backend/base/langflow/api/v1/validate.py +++ b/src/backend/base/langflow/api/v1/validate.py @@ -4,7 +4,7 @@ from fastapi import APIRouter, HTTPException from loguru import logger from langflow.api.v1.base import Code, CodeValidationResponse, PromptValidationResponse, ValidatePromptRequest -from langflow.base.prompts.utils import ( +from langflow.base.prompts.api_utils import ( add_new_variables_to_template, get_old_custom_fields, remove_old_variables_from_template, diff --git a/src/backend/base/langflow/base/prompts/api_utils.py b/src/backend/base/langflow/base/prompts/api_utils.py new file mode 100644 index 000000000..89a17399c --- /dev/null +++ b/src/backend/base/langflow/base/prompts/api_utils.py @@ -0,0 +1,83 @@ +from fastapi import HTTPException +from langchain.prompts import PromptTemplate +from loguru import logger + +from langflow.api.v1.base import INVALID_NAMES, check_input_variables +from langflow.interface.utils import extract_input_variables_from_prompt +from langflow.template.field.prompt import DefaultPromptField + + +def validate_prompt(prompt_template: str, silent_errors: bool = False) -> list[str]: + input_variables = extract_input_variables_from_prompt(prompt_template) + + # Check if there are invalid characters in the input_variables + input_variables = check_input_variables(input_variables) + if any(var in INVALID_NAMES for var in input_variables): + raise ValueError(f"Invalid input variables. None of the variables can be named {', '.join(input_variables)}. ") + + try: + PromptTemplate(template=prompt_template, input_variables=input_variables) + except Exception as exc: + logger.error(f"Invalid prompt: {exc}") + if not silent_errors: + raise ValueError(f"Invalid prompt: {exc}") from exc + + return input_variables + + +def get_old_custom_fields(custom_fields, name): + try: + if len(custom_fields) == 1 and name == "": + # If there is only one custom field and the name is empty string + # then we are dealing with the first prompt request after the node was created + name = list(custom_fields.keys())[0] + + old_custom_fields = custom_fields[name] + if not old_custom_fields: + old_custom_fields = [] + + old_custom_fields = old_custom_fields.copy() + except KeyError: + old_custom_fields = [] + custom_fields[name] = [] + return old_custom_fields + + +def add_new_variables_to_template(input_variables, custom_fields, template, name): + for variable in input_variables: + try: + template_field = DefaultPromptField(name=variable, display_name=variable) + if variable in template: + # Set the new field with the old value + template_field.value = template[variable]["value"] + + template[variable] = template_field.to_dict() + + # Check if variable is not already in the list before appending + if variable not in custom_fields[name]: + custom_fields[name].append(variable) + + except Exception as exc: + logger.exception(exc) + raise HTTPException(status_code=500, detail=str(exc)) from exc + + +def remove_old_variables_from_template(old_custom_fields, input_variables, custom_fields, template, name): + for variable in old_custom_fields: + if variable not in input_variables: + try: + # Remove the variable from custom_fields associated with the given name + if variable in custom_fields[name]: + custom_fields[name].remove(variable) + + # Remove the variable from the template + template.pop(variable, None) + + except Exception as exc: + logger.exception(exc) + raise HTTPException(status_code=500, detail=str(exc)) from exc + + +def update_input_variables_field(input_variables, template): + if "input_variables" in template: + template["input_variables"]["value"] = input_variables diff --git a/src/backend/base/langflow/base/prompts/utils.py b/src/backend/base/langflow/base/prompts/utils.py index ff4cf5079..fbf46dd71 100644 --- a/src/backend/base/langflow/base/prompts/utils.py +++ b/src/backend/base/langflow/base/prompts/utils.py @@ -1,12 +1,19 @@ -from fastapi import HTTPException -from langchain.prompts import PromptTemplate from langchain_core.documents import Document -from loguru import logger -from langflow.api.v1.base import INVALID_NAMES, check_input_variables -from langflow.interface.utils import extract_input_variables_from_prompt from langflow.schema import Record -from langflow.template.field.prompt import DefaultPromptField + + +def record_to_string(record: Record) -> str: + """ + Convert a record to a string. + + Args: + record (Record): The record to convert. + + Returns: + str: The record as a string. + """ + return record.get_text() def dict_values_to_string(d: dict) -> dict: @@ -35,19 +42,6 @@ def dict_values_to_string(d: dict) -> dict: return d -def record_to_string(record: Record) -> str: - """ - Convert a record to a string. - - Args: - record (Record): The record to convert. - - Returns: - str: The record as a string. - """ - return record.get_text() - - def document_to_string(document: Document) -> str: """ Convert a document to a string. @@ -59,79 +53,3 @@ def document_to_string(document: Document) -> str: str: The document as a string. """ return document.page_content - - -def validate_prompt(prompt_template: str, silent_errors: bool = False) -> list[str]: - input_variables = extract_input_variables_from_prompt(prompt_template) - - # Check if there are invalid characters in the input_variables - input_variables = check_input_variables(input_variables) - if any(var in INVALID_NAMES for var in input_variables): - raise ValueError(f"Invalid input variables. None of the variables can be named {', '.join(input_variables)}. ") - - try: - PromptTemplate(template=prompt_template, input_variables=input_variables) - except Exception as exc: - logger.error(f"Invalid prompt: {exc}") - if not silent_errors: - raise ValueError(f"Invalid prompt: {exc}") from exc - - return input_variables - - -def get_old_custom_fields(custom_fields, name): - try: - if len(custom_fields) == 1 and name == "": - # If there is only one custom field and the name is empty string - # then we are dealing with the first prompt request after the node was created - name = list(custom_fields.keys())[0] - - old_custom_fields = custom_fields[name] - if not old_custom_fields: - old_custom_fields = [] - - old_custom_fields = old_custom_fields.copy() - except KeyError: - old_custom_fields = [] - custom_fields[name] = [] - return old_custom_fields - - -def add_new_variables_to_template(input_variables, custom_fields, template, name): - for variable in input_variables: - try: - template_field = DefaultPromptField(name=variable, display_name=variable) - if variable in template: - # Set the new field with the old value - template_field.value = template[variable]["value"] - - template[variable] = template_field.to_dict() - - # Check if variable is not already in the list before appending - if variable not in custom_fields[name]: - custom_fields[name].append(variable) - - except Exception as exc: - logger.exception(exc) - raise HTTPException(status_code=500, detail=str(exc)) from exc - - -def remove_old_variables_from_template(old_custom_fields, input_variables, custom_fields, template, name): - for variable in old_custom_fields: - if variable not in input_variables: - try: - # Remove the variable from custom_fields associated with the given name - if variable in custom_fields[name]: - custom_fields[name].remove(variable) - - # Remove the variable from the template - template.pop(variable, None) - - except Exception as exc: - logger.exception(exc) - raise HTTPException(status_code=500, detail=str(exc)) from exc - - -def update_input_variables_field(input_variables, template): - if "input_variables" in template: - template["input_variables"]["value"] = input_variables diff --git a/tests/test_loading.py b/tests/test_loading.py index 8731e4b68..bea558f30 100644 --- a/tests/test_loading.py +++ b/tests/test_loading.py @@ -1,4 +1,5 @@ import pytest + from langflow.graph import Graph from langflow.graph.schema import RunOutputs from langflow.initial_setup.setup import load_starter_projects @@ -36,7 +37,7 @@ def test_load_flow_from_json_object(): def test_run_flow_from_json_object(): """Test loading a flow from a json file and applying tweaks""" _, projects = zip(*load_starter_projects()) - project = projects[0] + project = [project for project in projects if "Basic Prompting" in project["name"]][0] results = run_flow_from_json(project, input_value="test") assert results is not None assert all(isinstance(result, RunOutputs) for result in results)