Merge branch 'cz/mergeAll' into two_edges

src/backend/base/langflow/api/utils.py

@@ -219,7 +219,7 @@ async def build_and_cache_graph_from_db(flow_id: str, session: Session, chat_ser
         if vertex is None:
             raise ValueError(f"Vertex {vertex_id} not found")
         if not vertex._raw_params.get("session_id"):
-            vertex.update_raw_params({"session_id": flow_id})
+            vertex.update_raw_params({"session_id": flow_id}, overwrite=True)
     await chat_service.set_cache(flow_id, graph)
     return graph
 

src/backend/base/langflow/base/vectorstores/utils.py

@@ -0,0 +1,24 @@
+from langflow.schema import Record
+
+
+def chroma_collection_to_records(collection_dict: dict):
+    """
+    Converts a collection of chroma vectors into a list of records.
+
+    Args:
+        collection_dict (dict): A dictionary containing the collection of chroma vectors.
+
+    Returns:
+        list: A list of records, where each record represents a document in the collection.
+    """
+    records = []
+    for i, doc in enumerate(collection_dict["documents"]):
+        record_dict = {
+            "id": collection_dict["ids"][i],
+            "text": doc,
+        }
+        if "metadatas" in collection_dict:
+            for key, value in collection_dict["metadatas"][i].items():
+                record_dict[key] = value
+        records.append(Record(**record_dict))
+    return records

src/backend/base/langflow/components/outputs/RecordsOutput.py

@@ -15,4 +15,5 @@ class RecordsOutput(Component):
     ]
 
     def record_response(self) -> Record:
+        self.status = self.input_value
         return self.input_value

src/backend/base/langflow/components/retrievers/SelfQueryRetriever.py

@@ -0,0 +1,68 @@
+# from langflow.field_typing import Data
+from langchain.chains.query_constructor.base import AttributeInfo
+from langchain.retrievers.self_query.base import SelfQueryRetriever
+from langchain_core.vectorstores import VectorStore
+
+from langflow.custom import CustomComponent
+from langflow.field_typing import BaseLanguageModel, Text
+from langflow.schema import Record
+from langflow.schema.message import Message
+
+
+class SelfQueryRetrieverComponent(CustomComponent):
+    display_name: str = "Self Query Retriever"
+    description: str = "Retriever that uses a vector store and an LLM to generate the vector store queries."
+    icon = "LangChain"
+
+    def build_config(self):
+        return {
+            "query": {
+                "display_name": "Query",
+                "input_types": ["Message", "Text"],
+                "info": "Query to be passed as input.",
+            },
+            "vectorstore": {
+                "display_name": "Vector Store",
+                "info": "Vector Store to be passed as input.",
+            },
+            "attribute_infos": {
+                "display_name": "Metadata Field Info",
+                "info": "Metadata Field Info to be passed as input.",
+            },
+            "document_content_description": {
+                "display_name": "Document Content Description",
+                "info": "Document Content Description to be passed as input.",
+            },
+            "llm": {
+                "display_name": "LLM",
+                "info": "LLM to be passed as input.",
+            },
+        }
+
+    def build(
+        self,
+        query: Message,
+        vectorstore: VectorStore,
+        attribute_infos: list[Record],
+        document_content_description: Text,
+        llm: BaseLanguageModel,
+    ) -> Record:
+        metadata_field_infos = [AttributeInfo(**record.data) for record in attribute_infos]
+        self_query_retriever = SelfQueryRetriever.from_llm(
+            llm=llm,
+            vectorstore=vectorstore,
+            document_contents=document_content_description,
+            metadata_field_info=metadata_field_infos,
+            enable_limit=True,
+        )
+
+        if isinstance(query, Message):
+            input_text = query.text
+        elif isinstance(query, str):
+            input_text = query
+        else:
+            raise ValueError(f"Query type {type(query)} not supported.")
+        documents = self_query_retriever.invoke(input=input_text)
+        records = [Record.from_document(document) for document in documents]
+        self.status = records
+        return records

src/backend/base/langflow/components/vectorstores/Chroma.py

@@ -1,3 +1,4 @@
+from copy import deepcopy
 from typing import List, Optional, Union
 
 import chromadb
@@ -7,6 +8,7 @@ from langchain_core.embeddings import Embeddings
 from langchain_core.retrievers import BaseRetriever
 from langchain_core.vectorstores import VectorStore
 
+from langflow.base.vectorstores.utils import chroma_collection_to_records
 from langflow.custom import CustomComponent
 from langflow.schema import Record
 
@@ -48,6 +50,11 @@ class ChromaComponent(CustomComponent):
                 "display_name": "Server SSL Enabled",
                 "advanced": True,
             },
+            "allow_duplicates": {
+                "display_name": "Allow Duplicates",
+                "advanced": True,
+                "info": "If false, will not add documents that are already in the Vector Store.",
+            },
         }
 
     def build(
@@ -61,6 +68,7 @@ class ChromaComponent(CustomComponent):
         chroma_server_host: Optional[str] = None,
         chroma_server_http_port: Optional[int] = None,
         chroma_server_grpc_port: Optional[int] = None,
+        allow_duplicates: bool = False,
     ) -> Union[VectorStore, BaseRetriever]:
         """
         Builds the Vector Store or BaseRetriever object.
@@ -75,6 +83,7 @@ class ChromaComponent(CustomComponent):
         - chroma_server_host (Optional[str]): The host for the Chroma server.
         - chroma_server_http_port (Optional[int]): The HTTP port for the Chroma server.
         - chroma_server_grpc_port (Optional[int]): The gRPC port for the Chroma server.
+        - allow_duplicates (bool): Whether to allow duplicates in the Vector Store.
 
         Returns:
         - Union[VectorStore, BaseRetriever]: The Vector Store or BaseRetriever object.
@@ -93,32 +102,34 @@ class ChromaComponent(CustomComponent):
             )
             client = chromadb.HttpClient(settings=chroma_settings)
 
-        # If documents, then we need to create a Chroma instance using .from_documents
-
         # Check index_directory and expand it if it is a relative path
         if index_directory is not None:
             index_directory = self.resolve_path(index_directory)
 
+        chroma = Chroma(
+            persist_directory=index_directory,
+            client=client,
+            embedding_function=embedding,
+            collection_name=collection_name,
+        )
+        if allow_duplicates:
+            stored_records = []
+        else:
+            stored_records = chroma_collection_to_records(chroma.get())
+            _stored_documents_without_id = []
+            for record in deepcopy(stored_records):
+                del record.id
+                _stored_documents_without_id.append(record)
         documents = []
         for _input in inputs or []:
             if isinstance(_input, Record):
-                documents.append(_input.to_lc_document())
+                if _input not in _stored_documents_without_id:
+                    documents.append(_input.to_lc_document())
             else:
-                documents.append(_input)
-        if documents is not None and embedding is not None:
-            if len(documents) == 0:
-                raise ValueError("If documents are provided, there must be at least one document.")
-            chroma = Chroma.from_documents(
-                documents=documents,  # type: ignore
-                persist_directory=index_directory,
-                collection_name=collection_name,
-                embedding=embedding,
-                client=client,
-            )
-        else:
-            chroma = Chroma(
-                persist_directory=index_directory,
-                client=client,
-                embedding_function=embedding,
-            )
+                raise ValueError("Inputs must be a Record objects.")
+
+        if documents and embedding is not None:
+            chroma.add_documents(documents)
+
+        self.status = stored_records
         return chroma

src/backend/base/langflow/custom/custom_component/custom_component.py

@@ -123,8 +123,11 @@ class CustomComponent(BaseComponent):
     @staticmethod
     def resolve_path(path: str) -> str:
         """Resolves the path to an absolute path."""
+        if not path:
+            return path
         path_object = Path(path)
-        if path_object.parts[0] == "~":
+
+        if path_object.parts and path_object.parts[0] == "~":
             path_object = path_object.expanduser()
         elif path_object.is_relative_to("."):
             path_object = path_object.resolve()

src/backend/base/langflow/field_typing/prompt.py

@@ -25,6 +25,7 @@ class Prompt(Record):
         prompt_template = PromptTemplate.from_template(self.template)
         variables_with_str_values = dict_values_to_string(self.variables)
         formatted_prompt = prompt_template.format(**variables_with_str_values)
+        self.text = formatted_prompt
         return formatted_prompt
 
     @classmethod

src/backend/base/langflow/schema/record.py

@@ -1,12 +1,12 @@
 import copy
 import json
-from typing import cast, Optional
+from typing import Optional, cast
 
 from langchain_core.documents import Document
 from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
+from langchain_core.prompt_values import ImagePromptValue
 from langchain_core.prompts.image import ImagePromptTemplate
 from pydantic import BaseModel, model_serializer, model_validator
-from langchain_core.prompt_values import ImagePromptValue
 
 
 class Record(BaseModel):
@@ -200,3 +200,6 @@ class Record(BaseModel):
 
     def __contains__(self, key):
         return key in self.data
+
+    def __eq__(self, other):
+        return isinstance(other, Record) and self.data == other.data