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

Merge remote-tracking branch 'origin/dev' into celery

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-09-22 12:53:05 -03:00
commit 78ecb4c341
123 changed files with 4453 additions and 8987 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -55,7 +55,7 @@ def display_results(results):
def update_settings(
config: str,
cache: str,
cache: Optional[str] = None,
dev: bool = False,
remove_api_keys: bool = False,
components_path: Optional[Path] = None,
@ -153,10 +153,10 @@ def run(
log_file: Path = typer.Option(
"logs/langflow.log", help="Path to the log file.", envvar="LANGFLOW_LOG_FILE"
),
cache: str = typer.Option(
cache: Optional[str] = typer.Option(
envvar="LANGFLOW_LANGCHAIN_CACHE",
help="Type of cache to use. (InMemoryCache, SQLiteCache)",
default="SQLiteCache",
default=None,
),
jcloud: bool = typer.Option(False, help="Deploy on Jina AI Cloud"),
dev: bool = typer.Option(False, help="Run in development mode (may contain bugs)"),

49
src/backend/langflow/alembic/versions/67cc006d50bf_add_profile_image_column.py Normal file
View file

@ -0,0 +1,49 @@
"""Add profile-image column
Revision ID: 67cc006d50bf
Revises: 260dbcc8b680
Create Date: 2023-09-08 07:36:13.387318
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
from sqlalchemy.engine.reflection import Inspector
# revision identifiers, used by Alembic.
revision: str = "67cc006d50bf"
down_revision: Union[str, None] = "260dbcc8b680"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
conn = op.get_bind()
inspector = Inspector.from_engine(conn)
if "user" in inspector.get_table_names() and "profile_image" not in [
column["name"] for column in inspector.get_columns("user")
]:
with op.batch_alter_table("user", schema=None) as batch_op:
batch_op.add_column(
sa.Column(
"profile_image", sqlmodel.sql.sqltypes.AutoString(), nullable=True
)
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
conn = op.get_bind()
inspector = Inspector.from_engine(conn)
if "user" in inspector.get_table_names() and "profile_image" in [
column["name"] for column in inspector.get_columns("user")
]:
with op.batch_alter_table("user", schema=None) as batch_op:
batch_op.drop_column("profile_image")
# ### end Alembic commands ###

61
src/backend/langflow/api/v1/users.py
View file

@ -18,15 +18,17 @@ from langflow.services.auth.utils import (
get_current_active_superuser,
get_current_active_user,
get_password_hash,
verify_password,
)
from langflow.services.database.models.user.crud import (
get_user_by_id,
update_user,
)
router = APIRouter(tags=["Users"])
router = APIRouter(tags=["Users"], prefix="/users")
@router.post("/user", response_model=UserRead, status_code=201)
@router.post("/", response_model=UserRead, status_code=201)
def add_user(
user: UserCreate,
session: Session = Depends(get_session),
@ -50,7 +52,7 @@ def add_user(
return new_user
@router.get("/user", response_model=UserRead)
@router.get("/whoami", response_model=UserRead)
def read_current_user(
current_user: User = Depends(get_current_active_user),
) -> User:
@ -60,7 +62,7 @@ def read_current_user(
return current_user
@router.get("/users", response_model=UsersResponse)
@router.get("/", response_model=UsersResponse)
def read_all_users(
skip: int = 0,
limit: int = 10,
@ -82,20 +84,61 @@ def read_all_users(
)
@router.patch("/user/{user_id}", response_model=UserRead)
@router.patch("/{user_id}", response_model=UserRead)
def patch_user(
user_id: UUID,
user: UserUpdate,
_: Session = Depends(get_current_active_user),
user_update: UserUpdate,
user: User = Depends(get_current_active_user),
session: Session = Depends(get_session),
) -> User:
"""
Update an existing user's data.
"""
return update_user(user_id, user, session)
if not user.is_superuser and user.id != user_id:
raise HTTPException(
status_code=403, detail="You don't have the permission to update this user"
)
if user_update.password:
raise HTTPException(
status_code=400, detail="You can't change your password here"
)
if user_db := get_user_by_id(session, user_id):
return update_user(user_db, user_update, session)
else:
raise HTTPException(status_code=404, detail="User not found")
@router.delete("/user/{user_id}")
@router.patch("/{user_id}/reset-password", response_model=UserRead)
def reset_password(
user_id: UUID,
user_update: UserUpdate,
user: User = Depends(get_current_active_user),
session: Session = Depends(get_session),
) -> User:
"""
Reset a user's password.
"""
if user_id != user.id:
raise HTTPException(
status_code=400, detail="You can't change another user's password"
)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if verify_password(user_update.password, user.password):
raise HTTPException(
status_code=400, detail="You can't use your current password"
)
new_password = get_password_hash(user_update.password)
user.password = new_password
session.commit()
session.refresh(user)
return user
@router.delete("/{user_id}", response_model=dict)
def delete_user(
user_id: UUID,
current_user: User = Depends(get_current_active_superuser),

10
src/backend/langflow/api/v1/validate.py
View file

@ -58,6 +58,16 @@ def post_validate_prompt(prompt_request: ValidatePromptRequest):
def get_old_custom_fields(prompt_request):
try:
if (
len(prompt_request.frontend_node.custom_fields) == 1
and prompt_request.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
prompt_request.name = list(
prompt_request.frontend_node.custom_fields.keys()
)[0]
old_custom_fields = prompt_request.frontend_node.custom_fields[
prompt_request.name
].copy()

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

@ -42,8 +42,8 @@ class ConversationalAgent(CustomComponent):
self,
model_name: str,
openai_api_key: str,
openai_api_base: str,
tools: Tool,
openai_api_base: Optional[str] = None,
memory: Optional[BaseMemory] = None,
system_message: Optional[SystemMessagePromptTemplate] = None,
max_token_limit: int = 2000,

7
src/backend/langflow/components/chains/PromptRunner.py
View file

@ -16,17 +16,14 @@ class PromptRunner(CustomComponent):
"info": "Make sure the prompt has all variables filled.",
},
"code": {"show": False},
"inputs": {"field_type": "code"},
}
def build(
self,
llm: BaseLLM,
prompt: PromptTemplate,
self, llm: BaseLLM, prompt: PromptTemplate, inputs: dict = {}
) -> Document:
chain = prompt | llm
# The input is an empty dict because the prompt is already filled
result = chain.invoke({})
result = chain.invoke(input=inputs)
if hasattr(result, "content"):
result = result.content
self.repr_value = result

42
src/backend/langflow/components/llms/HuggingFaceEndpoints.py Normal file
View file

@ -0,0 +1,42 @@
from typing import Optional
from langflow import CustomComponent
from langchain.llms import HuggingFaceEndpoint
from langchain.llms.base import BaseLLM
class HuggingFaceEndpointsComponent(CustomComponent):
display_name: str = "Hugging Face Inference API"
description: str = "LLM model from Hugging Face Inference API."
def build_config(self):
return {
"endpoint_url": {"display_name": "Endpoint URL", "password": True},
"task": {
"display_name": "Task",
"type": "select",
"options": ["text2text-generation", "text-generation", "summarization"],
},
"huggingfacehub_api_token": {"display_name": "API token", "password": True},
"model_kwargs": {
"display_name": "Model Keyword Arguments",
"field_type": "code",
},
"code": {"show": False},
}
def build(
self,
endpoint_url: str,
task="text2text-generation",
huggingfacehub_api_token: Optional[str] = None,
model_kwargs: Optional[dict] = None,
) -> BaseLLM:
try:
output = HuggingFaceEndpoint(
endpoint_url=endpoint_url,
task=task,
huggingfacehub_api_token=huggingfacehub_api_token,
)
except Exception as e:
raise ValueError("Could not connect to HuggingFace Endpoints API.") from e
return output

0
src/backend/langflow/components/llms/__init__.py Normal file
View file

28
src/backend/langflow/components/retrievers/MetalRetriever.py Normal file
View file

@ -0,0 +1,28 @@
from typing import Optional
from langflow import CustomComponent
from langchain.retrievers import MetalRetriever
from langchain.schema import BaseRetriever
from metal_sdk.metal import Metal # type: ignore
class MetalRetrieverComponent(CustomComponent):
display_name: str = "Metal Retriever"
description: str = "Retriever that uses the Metal API."
def build_config(self):
return {
"api_key": {"display_name": "API Key", "password": True},
"client_id": {"display_name": "Client ID", "password": True},
"index_id": {"display_name": "Index ID"},
"params": {"display_name": "Parameters"},
"code": {"show": False},
}
def build(
self, api_key: str, client_id: str, index_id: str, params: Optional[dict] = None
) -> BaseRetriever:
try:
metal = Metal(api_key=api_key, client_id=client_id, index_id=index_id)
except Exception as e:
raise ValueError("Could not connect to Metal API.") from e
return MetalRetriever(client=metal, params=params or {})

0
src/backend/langflow/components/retrievers/__init__.py Normal file
View file

80
src/backend/langflow/components/textsplitters/LanguageRecursiveTextSplitter.py Normal file
View file

@ -0,0 +1,80 @@
from typing import Optional
from langflow import CustomComponent
from langchain.text_splitter import Language
from langchain.schema import Document
class LanguageRecursiveTextSplitterComponent(CustomComponent):
display_name: str = "Language Recursive Text Splitter"
description: str = "Split text into chunks of a specified length based on language."
documentation: str = "https://docs.langflow.org/components/text-splitters#languagerecursivetextsplitter"
def build_config(self):
options = [x.value for x in Language]
return {
"documents": {
"display_name": "Documents",
"info": "The documents to split.",
},
"separator_type": {
"display_name": "Separator Type",
"info": "The type of separator to use.",
"field_type": "str",
"options": options,
"value": "Python",
},
"separators": {
"display_name": "Separators",
"info": "The characters to split on.",
"is_list": True,
},
"chunk_size": {
"display_name": "Chunk Size",
"info": "The maximum length of each chunk.",
"field_type": "int",
"value": 1000,
},
"chunk_overlap": {
"display_name": "Chunk Overlap",
"info": "The amount of overlap between chunks.",
"field_type": "int",
"value": 200,
},
"code": {"show": False},
}
def build(
self,
documents: list[Document],
chunk_size: Optional[int] = 1000,
chunk_overlap: Optional[int] = 200,
separator_type: Optional[str] = "Python",
) -> list[Document]:
"""
Split text into chunks of a specified length.
Args:
separators (list[str]): The characters to split on.
chunk_size (int): The maximum length of each chunk.
chunk_overlap (int): The amount of overlap between chunks.
length_function (function): The function to use to calculate the length of the text.
Returns:
list[str]: The chunks of text.
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Make sure chunk_size and chunk_overlap are ints
if isinstance(chunk_size, str):
chunk_size = int(chunk_size)
if isinstance(chunk_overlap, str):
chunk_overlap = int(chunk_overlap)
splitter = RecursiveCharacterTextSplitter.from_language(
language=Language(separator_type),
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
)
docs = splitter.split_documents(documents)
return docs

79
src/backend/langflow/components/textsplitters/RecursiveCharacterTextSplitter.py Normal file
View file

@ -0,0 +1,79 @@
from typing import Optional
from langflow import CustomComponent
from langchain.schema import Document
from langflow.utils.util import build_loader_repr_from_documents
class RecursiveCharacterTextSplitterComponent(CustomComponent):
display_name: str = "Recursive Character Text Splitter"
description: str = "Split text into chunks of a specified length."
documentation: str = "https://docs.langflow.org/components/text-splitters#recursivecharactertextsplitter"
def build_config(self):
return {
"documents": {
"display_name": "Documents",
"info": "The documents to split.",
},
"separators": {
"display_name": "Separators",
"info": 'The characters to split on.\nIf left empty defaults to ["\\n\\n", "\\n", " ", ""].',
"is_list": True,
},
"chunk_size": {
"display_name": "Chunk Size",
"info": "The maximum length of each chunk.",
"field_type": "int",
"value": 1000,
},
"chunk_overlap": {
"display_name": "Chunk Overlap",
"info": "The amount of overlap between chunks.",
"field_type": "int",
"value": 200,
},
"code": {"show": False},
}
def build(
self,
documents: list[Document],
separators: Optional[list[str]] = None,
chunk_size: Optional[int] = 1000,
chunk_overlap: Optional[int] = 200,
) -> list[Document]:
"""
Split text into chunks of a specified length.
Args:
separators (list[str]): The characters to split on.
chunk_size (int): The maximum length of each chunk.
chunk_overlap (int): The amount of overlap between chunks.
length_function (function): The function to use to calculate the length of the text.
Returns:
list[str]: The chunks of text.
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
if separators == "":
separators = None
elif separators:
# check if the separators list has escaped characters
# if there are escaped characters, unescape them
separators = [x.encode().decode("unicode-escape") for x in separators]
# Make sure chunk_size and chunk_overlap are ints
if isinstance(chunk_size, str):
chunk_size = int(chunk_size)
if isinstance(chunk_overlap, str):
chunk_overlap = int(chunk_overlap)
splitter = RecursiveCharacterTextSplitter(
separators=separators,
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
)
docs = splitter.split_documents(documents)
self.repr_value = build_loader_repr_from_documents(docs)
return docs

0
src/backend/langflow/components/textsplitters/__init__.py Normal file
View file

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

@ -19,7 +19,6 @@ class GetRequest(CustomComponent):
},
"headers": {
"display_name": "Headers",
"field_type": "code",
"info": "The headers to send with the request.",
},
"code": {"show": False},

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

@ -15,7 +15,6 @@ class PostRequest(CustomComponent):
"url": {"display_name": "URL", "info": "The URL to make the request to."},
"headers": {
"display_name": "Headers",
"field_type": "code",
"info": "The headers to send with the request.",
},
"code": {"show": False},

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

@ -15,7 +15,7 @@ class UpdateRequest(CustomComponent):
"url": {"display_name": "URL", "info": "The URL to make the request to."},
"headers": {
"display_name": "Headers",
"field_type": "code",
"field_type": "NestedDict",
"info": "The headers to send with the request.",
},
"code": {"show": False},

2
src/backend/langflow/config.yaml
View file

@ -171,8 +171,6 @@ prompts:
textsplitters:
CharacterTextSplitter:
documentation: "https://python.langchain.com/docs/modules/data_connection/document_transformers/text_splitters/character_text_splitter"
RecursiveCharacterTextSplitter:
documentation: "https://python.langchain.com/docs/modules/data_connection/document_transformers/text_splitters/recursive_text_splitter"
toolkits:
OpenAPIToolkit:
documentation: ""

3
src/backend/langflow/field_typing/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .base import NestedDict
__all__ = ["NestedDict"]

4
src/backend/langflow/field_typing/base.py Normal file
View file

@ -0,0 +1,4 @@
from typing import Union, Dict
# Type alias for more complex dicts
NestedDict = Dict[str, Union[str, Dict]]

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

@ -195,6 +195,19 @@ class Vertex:
except Exception as exc:
logger.debug(f"Error parsing code: {exc}")
params[key] = value.get("value")
elif value.get("type") in ["dict", "NestedDict"]:
# When dict comes from the frontend it comes as a
# list of dicts, so we need to convert it to a dict
# before passing it to the build method
_value = value.get("value")
if isinstance(_value, list):
params[key] = {
k: v
for item in value.get("value", [])
for k, v in item.items()
}
elif isinstance(_value, dict):
params[key] = _value
else:
params[key] = value.get("value")

8
src/backend/langflow/interface/run.py
View file

@ -45,8 +45,12 @@ def get_memory_key(langchain_object):
"chat_history": "history",
"history": "chat_history",
}
memory_key = langchain_object.memory.memory_key
return mem_key_dict.get(memory_key)
# Check if memory_key attribute exists
if hasattr(langchain_object.memory, "memory_key"):
memory_key = langchain_object.memory.memory_key
return mem_key_dict.get(memory_key)
else:
return None # or some other default value or action
def update_memory_keys(langchain_object, possible_new_mem_key):

11
src/backend/langflow/interface/tools/util.py
View file

@ -1,5 +1,6 @@
import ast
import inspect
import textwrap
from typing import Dict, Union
from langchain.agents.tools import Tool
@ -7,7 +8,7 @@ from loguru import logger
def get_func_tool_params(func, **kwargs) -> Union[Dict, None]:
tree = ast.parse(inspect.getsource(func))
tree = ast.parse(textwrap.dedent(inspect.getsource(func)))
# Iterate over the statements in the abstract syntax tree
for node in ast.walk(tree):
@ -58,13 +59,7 @@ def get_func_tool_params(func, **kwargs) -> Union[Dict, None]:
def get_class_tool_params(cls, **kwargs) -> Union[Dict, None]:
try:
tree = ast.parse(inspect.getsource(cls))
except IndentationError:
logger.error(
f"Error parsing class {cls.__name__}. Make sure there are no tabs in the code."
)
return None
tree = ast.parse(textwrap.dedent(inspect.getsource(cls)))
tool_params = {}

67
src/backend/langflow/interface/types.py
View file

@ -5,6 +5,7 @@ from langflow.api.utils import merge_nested_dicts_with_renaming
from langflow.interface.agents.base import agent_creator
from langflow.interface.chains.base import chain_creator
from langflow.interface.custom.constants import CUSTOM_COMPONENT_SUPPORTED_TYPES
from langflow.interface.custom.utils import extract_inner_type
from langflow.interface.document_loaders.base import documentloader_creator
from langflow.interface.embeddings.base import embedding_creator
from langflow.interface.importing.utils import get_function_custom
@ -84,6 +85,8 @@ def build_langchain_types_dict(): # sourcery skip: dict-assign-update-to-union
def process_type(field_type: str):
if field_type.startswith("list") or field_type.startswith("List"):
return extract_inner_type(field_type)
return "prompt" if field_type == "Prompt" else field_type
@ -100,6 +103,7 @@ def add_new_custom_field(
# if it is, update the value
display_name = field_config.pop("display_name", field_name)
field_type = field_config.pop("field_type", field_type)
field_contains_list = "list" in field_type.lower()
field_type = process_type(field_type)
field_value = field_config.pop("value", field_value)
field_advanced = field_config.pop("advanced", False)
@ -110,7 +114,9 @@ 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)
field_config["is_list"] = (
is_list or field_config.get("is_list", False) or field_contains_list
)
if "name" in field_config:
warnings.warn(
@ -172,7 +178,7 @@ def extract_type_from_optional(field_type):
Returns:
str: The extracted type, or an empty string if no type was found.
"""
match = re.search(r"\[(.*?)\]", field_type)
match = re.search(r"\[(.*?)\]$", field_type)
return match[1] if match else None
@ -284,31 +290,42 @@ def add_base_classes(frontend_node, return_types: List[str]):
def build_langchain_template_custom_component(custom_component: CustomComponent):
"""Build a custom component template for the langchain"""
logger.debug("Building custom component template")
frontend_node = build_frontend_node(custom_component)
try:
logger.debug("Building custom component template")
frontend_node = build_frontend_node(custom_component)
if frontend_node is None:
return None
logger.debug("Built base frontend node")
template_config = custom_component.build_template_config
if frontend_node is None:
return None
logger.debug("Built base frontend node")
template_config = custom_component.build_template_config
update_attributes(frontend_node, template_config)
logger.debug("Updated attributes")
field_config = build_field_config(custom_component)
logger.debug("Built field config")
add_extra_fields(
frontend_node, field_config, custom_component.get_function_entrypoint_args
)
logger.debug("Added extra fields")
frontend_node = add_code_field(
frontend_node, custom_component.code, field_config.get("code", {})
)
logger.debug("Added code field")
add_base_classes(
frontend_node, custom_component.get_function_entrypoint_return_type
)
logger.debug("Added base classes")
return frontend_node
update_attributes(frontend_node, template_config)
logger.debug("Updated attributes")
field_config = build_field_config(custom_component)
logger.debug("Built field config")
add_extra_fields(
frontend_node, field_config, custom_component.get_function_entrypoint_args
)
logger.debug("Added extra fields")
frontend_node = add_code_field(
frontend_node, custom_component.code, field_config.get("code", {})
)
logger.debug("Added code field")
add_base_classes(
frontend_node, custom_component.get_function_entrypoint_return_type
)
logger.debug("Added base classes")
return frontend_node
except Exception as exc:
raise HTTPException(
status_code=400,
detail={
"error": (
"Invalid type convertion. Please check your code and try again."
),
"traceback": traceback.format_exc(),
},
) from exc
def load_files_from_path(path: str):

19
src/backend/langflow/interface/utils.py
View file

@ -77,11 +77,16 @@ def set_langchain_cache(settings):
import langchain
from langflow.interface.importing.utils import import_class
langchain_cache_type = os.getenv("LANGFLOW_LANGCHAIN_CACHE")
cache_class = import_class(
f"langchain.cache.{langchain_cache_type or settings.LANGCHAIN_CACHE}"
)
if cache_type := os.getenv("LANGFLOW_LANGCHAIN_CACHE"):
try:
cache_class = import_class(
f"langchain.cache.{cache_type or settings.LANGCHAIN_CACHE}"
)
logger.debug(f"Setting up LLM caching with {cache_class.__name__}")
langchain.llm_cache = cache_class()
logger.info(f"LLM caching setup with {cache_class.__name__}")
logger.debug(f"Setting up LLM caching with {cache_class.__name__}")
langchain.llm_cache = cache_class()
logger.info(f"LLM caching setup with {cache_class.__name__}")
except ImportError:
logger.warning(f"Could not import {cache_type}. ")
else:
logger.info("No LLM cache set.")

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

@ -11,6 +11,7 @@ from langflow.api import router
from langflow.interface.utils import setup_llm_caching
from langflow.services.database.utils import initialize_database
from langflow.services.manager import initialize_services, teardown_services
from langflow.services.plugins.langfuse import LangfuseInstance
from langflow.utils.logger import configure
@ -41,6 +42,8 @@ def create_app():
app.on_event("startup")(initialize_database)
app.on_event("startup")(setup_llm_caching)
app.on_event("shutdown")(teardown_services)
app.on_event("startup")(LangfuseInstance.update)
app.on_event("shutdown")(LangfuseInstance.teardown)
return app

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

@ -1,4 +1,4 @@
from typing import Union
from typing import List, Union, TYPE_CHECKING
from langflow.api.v1.callback import (
AsyncStreamingLLMCallbackHandler,
StreamingLLMCallbackHandler,
@ -6,6 +6,52 @@ from langflow.api.v1.callback import (
from langflow.processing.process import fix_memory_inputs, format_actions
from loguru import logger
from langchain.agents.agent import AgentExecutor
from langchain.callbacks.base import BaseCallbackHandler
if TYPE_CHECKING:
from langfuse.callback import CallbackHandler # type: ignore
def setup_callbacks(sync, trace_id, **kwargs):
"""Setup callbacks for langchain object"""
callbacks = []
if sync:
callbacks.append(StreamingLLMCallbackHandler(**kwargs))
else:
callbacks.append(AsyncStreamingLLMCallbackHandler(**kwargs))
if langfuse_callback := get_langfuse_callback(trace_id=trace_id):
logger.debug("Langfuse callback loaded")
callbacks.append(langfuse_callback)
return callbacks
def get_langfuse_callback(trace_id):
from langflow.services.plugins.langfuse import LangfuseInstance
from langfuse.callback import CreateTrace
logger.debug("Initializing langfuse callback")
if langfuse := LangfuseInstance.get():
logger.debug("Langfuse credentials found")
try:
trace = langfuse.trace(CreateTrace(id=trace_id))
return trace.getNewHandler()
except Exception as exc:
logger.error(f"Error initializing langfuse callback: {exc}")
return None
def flush_langfuse_callback_if_present(
callbacks: List[Union[BaseCallbackHandler, "CallbackHandler"]]
):
"""
If langfuse callback is present, run callback.langfuse.flush()
"""
for callback in callbacks:
if hasattr(callback, "langfuse"):
callback.langfuse.flush()
break
async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwargs):
@ -27,13 +73,18 @@ async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwa
logger.error(f"Error fixing memory inputs: {exc}")
try:
async_callbacks = [AsyncStreamingLLMCallbackHandler(**kwargs)]
output = await langchain_object.acall(inputs, callbacks=async_callbacks)
trace_id = kwargs.pop("session_id", None)
callbacks = setup_callbacks(sync=False, trace_id=trace_id, **kwargs)
output = await langchain_object.acall(inputs, callbacks=callbacks)
except Exception as exc:
# make the error message more informative
logger.debug(f"Error: {str(exc)}")
sync_callbacks = [StreamingLLMCallbackHandler(**kwargs)]
output = langchain_object(inputs, callbacks=sync_callbacks)
trace_id = kwargs.pop("session_id", None)
callbacks = setup_callbacks(sync=True, trace_id=trace_id, **kwargs)
output = langchain_object(inputs, callbacks=callbacks)
# 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 []

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

@ -12,6 +12,7 @@ from langflow.graph import Graph
from langchain.chains.base import Chain
from langchain.vectorstores.base import VectorStore
from typing import Any, Dict, List, Optional, Tuple, Union
from langchain.schema import Document
from pydantic import BaseModel
@ -139,10 +140,11 @@ def generate_result(langchain_object: Union[Chain, VectorStore], inputs: dict):
logger.debug("Generated result and thought")
elif isinstance(langchain_object, VectorStore):
result = langchain_object.search(**inputs)
elif isinstance(langchain_object, Document):
result = langchain_object.dict()
else:
raise ValueError(
f"Unknown langchain_object type: {type(langchain_object).__name__}"
)
logger.warning(f"Unknown langchain_object type: {type(langchain_object)}")
result = langchain_object
return result

7
src/backend/langflow/services/chat/manager.py
View file

@ -1,4 +1,5 @@
from collections import defaultdict
import uuid
from fastapi import WebSocket, status
from langflow.api.v1.schemas import ChatMessage, ChatResponse, FileResponse
from langflow.interface.utils import pil_to_base64
@ -47,6 +48,7 @@ class ChatService(Service):
def __init__(self):
self.active_connections: Dict[str, WebSocket] = {}
self.connection_ids: Dict[str, str] = {}
self.chat_history = ChatHistory()
self.chat_cache = cache_service
self.chat_cache.attach(self.update)
@ -91,9 +93,13 @@ class ChatService(Service):
async def connect(self, client_id: str, websocket: WebSocket):
self.active_connections[client_id] = websocket
# This is to avoid having multiple clients with the same id
#! Temporary solution
self.connection_ids[client_id] = f"{client_id}-{uuid.uuid4()}"
def disconnect(self, client_id: str):
self.active_connections.pop(client_id, None)
self.connection_ids.pop(client_id, None)
async def send_message(self, client_id: str, message: str):
websocket = self.active_connections[client_id]
@ -135,6 +141,7 @@ class ChatService(Service):
langchain_object=langchain_object,
chat_inputs=chat_inputs,
websocket=self.active_connections[client_id],
session_id=self.connection_ids[client_id],
)
self.set_cache(client_id, langchain_object)
except Exception as e:

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

@ -9,6 +9,7 @@ async def process_graph(
langchain_object,
chat_inputs: ChatMessage,
websocket: WebSocket,
session_id: str,
):
langchain_object = try_setting_streaming_options(langchain_object, websocket)
logger.debug("Loaded langchain object")
@ -27,7 +28,10 @@ async def process_graph(
logger.debug("Generating result and thought")
result, intermediate_steps = await get_result_and_steps(
langchain_object, chat_inputs.message, websocket=websocket
langchain_object,
chat_inputs.message,
websocket=websocket,
session_id=session_id,
)
logger.debug("Generated result and intermediate_steps")
return result, intermediate_steps

24
src/backend/langflow/services/database/models/user/crud.py
View file

@ -1,12 +1,12 @@
from datetime import datetime, timezone
from typing import Union
from uuid import UUID
from fastapi import Depends, HTTPException
from fastapi import Depends, HTTPException, status
from langflow.services.database.models.user.user import User, UserUpdate
from langflow.services.utils import get_session
from sqlalchemy.exc import IntegrityError
from sqlmodel import Session
from typing import Optional
from sqlalchemy.orm.attributes import flag_modified
@ -20,20 +20,26 @@ def get_user_by_id(db: Session, id: UUID) -> Union[User, None]:
def update_user(
user_id: UUID, user: UserUpdate, db: Session = Depends(get_session)
user_db: Optional[User], user: UserUpdate, db: Session = Depends(get_session)
) -> User:
user_db = get_user_by_id(db, user_id)
if not user_db:
raise HTTPException(status_code=404, detail="User not found")
user_db_by_username = get_user_by_username(db, user.username) # type: ignore
if user_db_by_username and user_db_by_username.id != user_id:
raise HTTPException(status_code=409, detail="Username already exists")
# user_db_by_username = get_user_by_username(db, user.username) # type: ignore
# if user_db_by_username and user_db_by_username.id != user_id:
# raise HTTPException(status_code=409, detail="Username already exists")
user_data = user.dict(exclude_unset=True)
changed = False
for attr, value in user_data.items():
if hasattr(user_db, attr) and value is not None:
setattr(user_db, attr, value)
changed = True
if not changed:
raise HTTPException(
status_code=status.HTTP_304_NOT_MODIFIED, detail="Nothing to update"
)
user_db.updated_at = datetime.now(timezone.utc)
flag_modified(user_db, "updated_at")
@ -49,5 +55,5 @@ def update_user(
def update_user_last_login_at(user_id: UUID, db: Session = Depends(get_session)):
user_data = UserUpdate(last_login_at=datetime.now(timezone.utc)) # type: ignore
return update_user(user_id, user_data, db)
user = get_user_by_id(db, user_id)
return update_user(user, user_data, db)

5
src/backend/langflow/services/database/models/user/user.py
View file

@ -15,6 +15,7 @@ class User(SQLModelSerializable, table=True):
id: UUID = Field(default_factory=uuid4, primary_key=True, unique=True)
username: str = Field(index=True, unique=True)
password: str = Field()
profile_image: Optional[str] = Field(default=None)
is_active: bool = Field(default=False)
is_superuser: bool = Field(default=False)
create_at: datetime = Field(default_factory=datetime.utcnow)
@ -32,6 +33,7 @@ class UserCreate(SQLModel):
class UserRead(SQLModel):
id: UUID = Field(default_factory=uuid4)
username: str = Field()
profile_image: Optional[str] = Field()
is_active: bool = Field()
is_superuser: bool = Field()
create_at: datetime = Field()
@ -40,7 +42,8 @@ class UserRead(SQLModel):
class UserUpdate(SQLModel):
username: Optional[str] = Field()
profile_image: Optional[str] = Field()
password: Optional[str] = Field()
is_active: Optional[bool] = Field()
is_superuser: Optional[bool] = Field()
last_login_at: Optional[datetime] = Field()

0
src/backend/langflow/services/plugins/__init__.py Normal file
View file

50
src/backend/langflow/services/plugins/langfuse.py Normal file
View file

@ -0,0 +1,50 @@
from langflow.services.utils import get_settings_manager
from langflow.utils.logger import logger
### Temporary implementation
# This will be replaced by a plugin system once merged into 0.5.0
class LangfuseInstance:
_instance = None
@classmethod
def get(cls):
logger.debug("Getting Langfuse instance")
if cls._instance is None:
cls.create()
return cls._instance
@classmethod
def create(cls):
logger.debug("Creating Langfuse instance")
from langfuse import Langfuse # type: ignore
settings_manager = get_settings_manager()
if (
settings_manager.settings.LANGFUSE_PUBLIC_KEY
and settings_manager.settings.LANGFUSE_SECRET_KEY
):
logger.debug("Langfuse credentials found")
cls._instance = Langfuse(
public_key=settings_manager.settings.LANGFUSE_PUBLIC_KEY,
secret_key=settings_manager.settings.LANGFUSE_SECRET_KEY,
host=settings_manager.settings.LANGFUSE_HOST,
)
else:
logger.debug("No Langfuse credentials found")
cls._instance = None
@classmethod
def update(cls):
logger.debug("Updating Langfuse instance")
cls._instance = None
cls.create()
@classmethod
def teardown(cls):
logger.debug("Tearing down Langfuse instance")
if cls._instance is not None:
cls._instance.flush()
cls._instance = None

2
src/backend/langflow/services/settings/auth.py
View file

@ -30,7 +30,7 @@ class AuthSettings(BaseSettings):
# If AUTO_LOGIN = True
# > The application does not request login and logs in automatically as a super user.
AUTO_LOGIN: bool = False
AUTO_LOGIN: bool = True
FIRST_SUPERUSER: str = "langflow"
FIRST_SUPERUSER_PASSWORD: str = "langflow"

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

@ -48,6 +48,10 @@ class Settings(BaseSettings):
REDIS_DB: int = 0
REDIS_CACHE_EXPIRE: int = 3600
LANGFUSE_SECRET_KEY: Optional[str] = None
LANGFUSE_PUBLIC_KEY: Optional[str] = None
LANGFUSE_HOST: Optional[str] = None
@validator("CONFIG_DIR", pre=True, allow_reuse=True)
def set_langflow_dir(cls, value):
if not value:

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

@ -140,13 +140,16 @@ class FrontendNode(BaseModel):
@staticmethod
def handle_dict_type(field: TemplateField, _type: str) -> str:
"""Handles 'dict' type by replacing it with 'code' or 'file' based on the field name."""
if "dict" in _type.lower():
if field.name == "dict_":
field.field_type = "file"
field.suffixes = [".json", ".yaml", ".yml"]
field.file_types = ["json", "yaml", "yml"]
else:
field.field_type = "code"
if "dict" in _type.lower() and field.name == "dict_":
field.field_type = "file"
field.suffixes = [".json", ".yaml", ".yml"]
field.file_types = ["json", "yaml", "yml"]
elif (
_type.startswith("Dict")
or _type.startswith("Mapping")
or _type.startswith("dict")
):
field.field_type = "dict"
return _type
@staticmethod
@ -240,20 +243,6 @@ class FrontendNode(BaseModel):
"description",
}
@staticmethod
def replace_dict_with_code_or_file(
field: TemplateField, _type: str, key: str
) -> str:
"""Replaces 'dict' type with 'code' or 'file'."""
if "dict" in _type.lower():
if key == "dict_":
field.field_type = "file"
field.suffixes = [".json", ".yaml", ".yml"]
field.file_types = ["json", "yaml", "yml"]
else:
field.field_type = "code"
return field.field_type
@staticmethod
def set_field_default_value(field: TemplateField, value: dict, key: str) -> None:
"""Sets the field value with the default value if present."""

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

@ -170,11 +170,11 @@ class DocumentLoaderFrontNode(FrontendNode):
# add a metadata field of type dict
self.template.add_field(
TemplateField(
field_type="code",
field_type="dict",
required=True,
show=True,
name="metadata",
value="{}",
value={},
display_name="Metadata",
multiline=False,
)

2
src/backend/langflow/template/frontend_node/embeddings.py
View file

@ -89,7 +89,7 @@ class EmbeddingFrontendNode(FrontendNode):
if field.name == "headers":
field.show = False
if field.name == "model_kwargs":
field.field_type = "code"
field.field_type = "dict"
field.advanced = True
field.show = True
elif field.name in [

17
src/backend/langflow/template/frontend_node/formatter/field_formatters.py
View file

@ -153,10 +153,13 @@ class DictCodeFileFormatter(FieldFormatter):
key = field.name
value = field.to_dict()
_type = value["type"]
if "dict" in _type.lower():
if key == "dict_":
field.field_type = "file"
field.suffixes = [".json", ".yaml", ".yml"]
field.file_types = ["json", "yaml", "yml"]
else:
field.field_type = "code"
if "dict" in _type.lower() and key == "dict_":
field.field_type = "file"
field.suffixes = [".json", ".yaml", ".yml"]
field.file_types = ["json", "yaml", "yml"]
elif (
_type.startswith("Dict")
or _type.startswith("Mapping")
or _type.startswith("dict")
):
field.field_type = "dict"

2
src/backend/langflow/template/frontend_node/llms.py
View file

@ -131,7 +131,7 @@ class LLMFrontendNode(FrontendNode):
if display_name := display_names_dict.get(field.name):
field.display_name = display_name
if field.name == "model_kwargs":
field.field_type = "code"
field.field_type = "dict"
field.advanced = True
field.show = True
elif field.name in [

1
src/backend/langflow/template/frontend_node/prompts.py
View file

@ -15,6 +15,7 @@ from langflow.template.template.base import Template
class PromptFrontendNode(FrontendNode):
@staticmethod
def format_field(field: TemplateField, name: Optional[str] = None) -> None:
FrontendNode.format_field(field, name)
# if field.field_type == "StringPromptTemplate"
# change it to str
PROMPT_FIELDS = [

1
src/backend/langflow/template/frontend_node/utilities.py
View file

@ -21,5 +21,4 @@ class UtilitiesFrontendNode(FrontendNode):
field.field_type = "str"
if isinstance(field.value, dict):
field.field_type = "code"
field.value = orjson_dumps(field.value)

2
src/backend/langflow/template/frontend_node/vectorstores.py
View file

@ -56,7 +56,7 @@ class VectorStoreFrontendNode(FrontendNode):
# Add search_kwargs field
extra_field = TemplateField(
name="search_kwargs",
field_type="code",
field_type="NestedDict",
required=False,
placeholder="",
show=True,

13
src/backend/langflow/utils/constants.py
View file

@ -48,5 +48,16 @@ def python_function(text: str) -> str:
return text
"""
DIRECT_TYPES = ["str", "bool", "code", "int", "float", "Any", "prompt"]
PYTHON_BASIC_TYPES = [str, bool, int, float, tuple, list, dict, set]
DIRECT_TYPES = [
"str",
"bool",
"dict",
"int",
"float",
"Any",
"prompt",
"code",
"NestedDict",
]

22
src/backend/langflow/utils/util.py
View file

@ -2,12 +2,13 @@ import re
import inspect
import importlib
from functools import wraps
from typing import Optional, Dict, Any, Union
from typing import List, Optional, Dict, Any, Union
from docstring_parser import parse
from langflow.template.frontend_node.constants import FORCE_SHOW_FIELDS
from langflow.utils import constants
from langchain.schema import Document
def build_template_from_function(
@ -275,8 +276,6 @@ def format_dict(
value["password"] = is_password_field(key)
value["multiline"] = is_multiline_field(key)
replace_dict_type_with_code(value)
if key == "dict_":
set_dict_file_attributes(value)
@ -406,14 +405,6 @@ def is_multiline_field(key: str) -> bool:
}
def replace_dict_type_with_code(value: Dict[str, Any]) -> None:
"""
Replaces the type value with 'code' if the type is a dict.
"""
if "dict" in value["type"].lower():
value["type"] = "code"
def set_dict_file_attributes(value: Dict[str, Any]) -> None:
"""
Sets the file attributes for the 'dict_' key.
@ -456,3 +447,12 @@ def add_options_to_field(
value["options"] = options_map[class_name]
value["list"] = True
value["value"] = options_map[class_name][0]
def build_loader_repr_from_documents(documents: List[Document]) -> str:
if documents:
avg_length = sum(len(doc.page_content) for doc in documents) / len(documents)
return f"""{len(documents)} documents
\nAvg. Document Length (characters): {int(avg_length)}
Documents: {documents[:3]}..."""
return "0 documents"