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

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

Browse source
This commit is contained in:
ogabrielluiz 2024-06-10 21:07:02 -03:00
commit 6d51386b83
59 changed files with 1266 additions and 544 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/backend/base/langflow/base/tools/base.py
View file

@ -9,7 +9,7 @@ def build_status_from_tool(tool: Tool):
tool (Tool): The tool object to build the status for.
Returns:
str: The status string representation of the tool, including its name, description, and arguments (if any).
str: The status string representation of the tool, including its name, description, arguments (if any), and args_schema (if any).
"""
description_repr = repr(tool.description).strip("'")
args_str = "\n".join(
@ -19,5 +19,8 @@ def build_status_from_tool(tool: Tool):
if "description" in arg_data
]
)
# Include args_schema information
args_schema_str = repr(tool.args_schema) if tool.args_schema else "None"
status = f"Name: {tool.name}\nDescription: {description_repr}"
status += f"\nArgs Schema: {args_schema_str}"
return status + (f"\nArguments:\n{args_str}" if args_str else "")

15
src/backend/base/langflow/components/memories/AstraDBMessageReader.py
View file

@ -1,9 +1,6 @@
from typing import Optional, cast
from langchain_astradb.chat_message_histories import AstraDBChatMessageHistory
from langflow.base.memory.memory import BaseMemoryComponent
from langflow.field_typing import Text
from langflow.schema import Record
@ -51,6 +48,14 @@ class AstraDBMessageReaderComponent(BaseMemoryComponent):
Returns:
list[Record]: A list of Record objects representing the search results.
"""
try:
from langchain_astradb.chat_message_histories import AstraDBChatMessageHistory
except ImportError:
raise ImportError(
"Could not import langchain Astra DB integration package. "
"Please install it with `pip install langchain-astradb`."
)
memory: AstraDBChatMessageHistory = cast(AstraDBChatMessageHistory, kwargs.get("memory"))
if not memory:
raise ValueError("AstraDBChatMessageHistory instance is required.")
@ -63,14 +68,14 @@ class AstraDBMessageReaderComponent(BaseMemoryComponent):
def build(
self,
session_id: Text,
session_id: str,
collection_name: str,
token: str,
api_endpoint: str,
namespace: Optional[str] = None,
) -> list[Record]:
try:
pass
from langchain_astradb.chat_message_histories import AstraDBChatMessageHistory
except ImportError:
raise ImportError(
"Could not import langchain Astra DB integration package. "

28
src/backend/base/langflow/components/memories/AstraDBMessageWriter.py
View file

@ -1,10 +1,8 @@
from typing import Optional
from langchain_astradb import AstraDBChatMessageHistory
from langchain_core.messages import BaseMessage
from langflow.base.memory.memory import BaseMemoryComponent
from langflow.field_typing import Text
from langflow.schema import Record
@ -50,7 +48,7 @@ class AstraDBMessageWriterComponent(BaseMemoryComponent):
self,
sender: str,
sender_name: str,
text: Text,
text: str,
session_id: str,
metadata: Optional[dict] = None,
**kwargs,
@ -59,17 +57,27 @@ class AstraDBMessageWriterComponent(BaseMemoryComponent):
Adds a message to the AstraDBChatMessageHistory memory.
Args:
sender (Text): The type of the message sender. Valid values are "Machine" or "User".
sender_name (Text): The name of the message sender.
text (Text): The content of the message.
session_id (Text): The session ID associated with the message.
sender (str): The type of the message sender. Typically "ai" or "human".
sender_name (str): The name of the message sender.
text (str): The content of the message.
session_id (str): The session ID associated with the message.
metadata (dict | None, optional): Additional metadata for the message. Defaults to None.
**kwargs: Additional keyword arguments.
**kwargs: Additional keyword arguments, including:
memory (AstraDBChatMessageHistory | None): The memory instance to add the message to.
Raises:
ValueError: If the AstraDBChatMessageHistory instance is not provided.
"""
try:
from langchain_astradb.chat_message_histories import AstraDBChatMessageHistory
except ImportError:
raise ImportError(
"Could not import langchain Astra DB integration package. "
"Please install it with `pip install langchain-astradb`."
)
memory: AstraDBChatMessageHistory | None = kwargs.pop("memory", None)
if memory is None:
raise ValueError("AstraDBChatMessageHistory instance is required.")
@ -89,14 +97,14 @@ class AstraDBMessageWriterComponent(BaseMemoryComponent):
def build(
self,
input_value: Record,
session_id: Text,
session_id: str,
collection_name: str,
token: str,
api_endpoint: str,
namespace: Optional[str] = None,
) -> Record:
try:
pass
from langchain_astradb.chat_message_histories import AstraDBChatMessageHistory
except ImportError:
raise ImportError(
"Could not import langchain Astra DB integration package. "

86
src/backend/base/langflow/components/memories/CassandraMessageReader.py Normal file
View file

@ -0,0 +1,86 @@
from typing import Optional, cast
from langchain_community.chat_message_histories import CassandraChatMessageHistory
from langflow.base.memory.memory import BaseMemoryComponent
from langflow.schema.schema import Record
class CassandraMessageReaderComponent(BaseMemoryComponent):
display_name = "Cassandra Message Reader"
description = "Retrieves stored chat messages from a Cassandra table on Astra DB."
def build_config(self):
return {
"session_id": {
"display_name": "Session ID",
"info": "Session ID of the chat history.",
"input_types": ["Text"],
},
"database_id": {
"display_name": "Database ID",
"info": "The Astra database ID.",
},
"table_name": {
"display_name": "Table Name",
"info": "The name of the table where messages are stored.",
},
"token": {
"display_name": "Token",
"info": "Authentication token for accessing Cassandra on Astra DB.",
"password": True,
},
"keyspace": {
"display_name": "Keyspace",
"info": "Optional key space within Astra DB. The keyspace should already be created.",
"input_types": ["Text"],
"advanced": True,
},
}
def get_messages(self, **kwargs) -> list[Record]:
"""
Retrieves messages from the CassandraChatMessageHistory memory.
Args:
memory (CassandraChatMessageHistory): The CassandraChatMessageHistory instance to retrieve messages from.
Returns:
list[Record]: A list of Record objects representing the search results.
"""
memory: CassandraChatMessageHistory = cast(CassandraChatMessageHistory, kwargs.get("memory"))
if not memory:
raise ValueError("CassandraChatMessageHistory instance is required.")
# Get messages from the memory
messages = memory.messages
results = [Record.from_lc_message(message) for message in messages]
return list(results)
def build(
self,
session_id: str,
table_name: str,
token: str,
database_id: str,
keyspace: Optional[str] = None,
) -> list[Record]:
try:
import cassio
except ImportError:
raise ImportError(
"Could not import cassio integration package. " "Please install it with `pip install cassio`."
)
cassio.init(token=token, database_id=database_id)
memory = CassandraChatMessageHistory(
session_id=session_id,
table_name=table_name,
keyspace=keyspace,
)
records = self.get_messages(memory=memory)
self.status = records
return records

122
src/backend/base/langflow/components/memories/CassandraMessageWriter.py Normal file
View file

@ -0,0 +1,122 @@
from typing import Optional
from langflow.base.memory.memory import BaseMemoryComponent
from langflow.schema.schema import Record
from langchain_core.messages import BaseMessage
from langchain_community.chat_message_histories import CassandraChatMessageHistory
class CassandraMessageWriterComponent(BaseMemoryComponent):
display_name = "Cassandra Message Writer"
description = "Writes a message to a Cassandra table on Astra DB."
def build_config(self):
return {
"input_value": {
"display_name": "Input Record",
"info": "Record to write to Cassandra.",
},
"session_id": {
"display_name": "Session ID",
"info": "Session ID of the chat history.",
"input_types": ["Text"],
},
"database_id": {
"display_name": "Database ID",
"info": "The Astra database ID.",
},
"table_name": {
"display_name": "Table Name",
"info": "The name of the table where messages will be stored.",
},
"token": {
"display_name": "Token",
"info": "Authentication token for accessing Cassandra on Astra DB.",
"password": True,
},
"keyspace": {
"display_name": "Keyspace",
"info": "Optional key space within Astra DB. The keyspace should already be created.",
"input_types": ["Text"],
"advanced": True,
},
"ttl_seconds": {
"display_name": "TTL Seconds",
"info": "Optional time-to-live for the messages.",
"input_types": ["Number"],
"advanced": True,
},
}
def add_message(
self,
sender: str,
sender_name: str,
text: str,
session_id: str,
metadata: Optional[dict] = None,
**kwargs,
):
"""
Adds a message to the CassandraChatMessageHistory memory.
Args:
sender (str): The type of the message sender. Typically "ai" or "human".
sender_name (str): The name of the message sender.
text (str): The content of the message.
session_id (str): The session ID associated with the message.
metadata (dict | None, optional): Additional metadata for the message. Defaults to None.
**kwargs: Additional keyword arguments, including:
memory (CassandraChatMessageHistory | None): The memory instance to add the message to.
Raises:
ValueError: If the CassandraChatMessageHistory instance is not provided.
"""
memory: CassandraChatMessageHistory | None = kwargs.pop("memory", None)
if memory is None:
raise ValueError("CassandraChatMessageHistory instance is required.")
text_list = [
BaseMessage(
content=text,
sender=sender,
sender_name=sender_name,
metadata=metadata,
session_id=session_id,
)
]
memory.add_messages(text_list)
def build(
self,
input_value: Record,
session_id: str,
table_name: str,
token: str,
database_id: str,
keyspace: Optional[str] = None,
ttl_seconds: Optional[int] = None,
) -> Record:
try:
import cassio
except ImportError:
raise ImportError(
"Could not import cassio integration package. " "Please install it with `pip install cassio`."
)
cassio.init(token=token, database_id=database_id)
memory = CassandraChatMessageHistory(
session_id=session_id,
table_name=table_name,
keyspace=keyspace,
ttl_seconds=ttl_seconds,
)
self.add_message(**input_value.data, memory=memory)
self.status = f"Added message to Cassandra memory for session {session_id}"
return input_value

2
src/backend/base/langflow/components/model_specs/ChatOpenAISpecs.py
View file

@ -28,7 +28,7 @@ class ChatOpenAIComponent(CustomComponent):
"model_name": {"display_name": "Model Name", "advanced": False, "options": MODEL_NAMES},
"openai_api_base": {
"display_name": "OpenAI API Base",
"advanced": False,
"advanced": True,
"required": False,
"info": (
"The base URL of the OpenAI API. Defaults to https://api.openai.com/v1.\n\n"

94
src/backend/base/langflow/components/vectorsearch/CassandraSearch.py Normal file
View file

@ -0,0 +1,94 @@
from typing import Any, List, Optional, Tuple
from langflow.components.vectorstores.Cassandra import CassandraVectorStoreComponent
from langflow.components.vectorstores.base.model import LCVectorStoreComponent
from langflow.field_typing import Embeddings, Text
from langflow.schema import Record
from langchain_community.utilities.cassandra import SetupMode
class CassandraSearchComponent(LCVectorStoreComponent):
display_name = "Cassandra Search"
description = "Searches an existing Cassandra Vector Store."
icon = "Cassandra"
field_order = ["token", "database_id", "table_name", "input_value", "embedding"]
def build_config(self):
return {
"search_type": {
"display_name": "Search Type",
"options": ["Similarity", "MMR"],
},
"input_value": {
"display_name": "Input Value",
"info": "Input value to search",
},
"embedding": {"display_name": "Embedding", "info": "Embedding to use"},
"token": {
"display_name": "Token",
"info": "Authentication token for accessing Cassandra on Astra DB.",
"password": True,
},
"database_id": {
"display_name": "Database ID",
"info": "The Astra database ID.",
},
"table_name": {
"display_name": "Table Name",
"info": "The name of the table where vectors will be stored.",
},
"keyspace": {
"display_name": "Keyspace",
"info": "Optional key space within Astra DB. The keyspace should already be created.",
"advanced": True,
},
"body_index_options": {
"display_name": "Body Index Options",
"info": "Optional options used to create the body index.",
"advanced": True,
},
"setup_mode": {
"display_name": "Setup Mode",
"info": "Configuration mode for setting up the Cassandra table, with options like 'Sync', 'Async', or 'Off'.",
"options": ["Sync", "Async", "Off"],
"advanced": True,
},
"number_of_results": {
"display_name": "Number of Results",
"info": "Number of results to return.",
"advanced": True,
},
}
def build(
self,
embedding: Embeddings,
table_name: str,
input_value: Text,
token: str,
database_id: str,
search_type: str = "similarity",
number_of_results: int = 4,
keyspace: Optional[str] = None,
body_index_options: Optional[List[Tuple[str, Any]]] = None,
setup_mode: SetupMode = SetupMode.SYNC,
) -> List[Record]:
vector_store = CassandraVectorStoreComponent().build(
embedding=embedding,
table_name=table_name,
token=token,
database_id=database_id,
keyspace=keyspace,
body_index_options=body_index_options,
setup_mode=setup_mode,
)
try:
return self.search_with_vector_store(input_value, search_type, vector_store, k=number_of_results)
except KeyError as e:
if "content" in str(e):
raise ValueError(
"You should ingest data through Langflow (or LangChain) to query it in Langflow. Your collection does not contain a field name 'content'."
)
else:
raise e

11
src/backend/base/langflow/components/vectorstores/AstraDB.py
View file

@ -1,7 +1,5 @@
from typing import List, Optional, Union
from langchain_astradb import AstraDBVectorStore
from langchain_astradb.utils.astradb import SetupMode
from langchain_core.retrievers import BaseRetriever
from langflow.custom import CustomComponent
@ -112,6 +110,15 @@ class AstraDBVectorStoreComponent(CustomComponent):
metadata_indexing_exclude: Optional[List[str]] = None,
collection_indexing_policy: Optional[dict] = None,
) -> Union[VectorStore, BaseRetriever]:
try:
from langchain_astradb import AstraDBVectorStore
from langchain_astradb.utils.astradb import SetupMode
except ImportError:
raise ImportError(
"Could not import langchain Astra DB integration package. "
"Please install it with `pip install langchain-astradb`."
)
try:
setup_mode_value = SetupMode[setup_mode.upper()]
except KeyError:

110
src/backend/base/langflow/components/vectorstores/Cassandra.py Normal file
View file

@ -0,0 +1,110 @@
from typing import Any, List, Optional, Tuple
from langchain_community.vectorstores import Cassandra
from langchain_community.utilities.cassandra import SetupMode
from langflow.custom import CustomComponent
from langflow.field_typing import Embeddings, VectorStore
from langflow.schema import Record
class CassandraVectorStoreComponent(CustomComponent):
display_name = "Cassandra"
description = "Builds or loads a Cassandra Vector Store."
icon = "Cassandra"
field_order = ["token", "database_id", "table_name", "inputs", "embedding"]
def build_config(self):
return {
"inputs": {
"display_name": "Inputs",
"info": "Optional list of records to be processed and stored in the vector store.",
},
"embedding": {"display_name": "Embedding", "info": "Embedding to use"},
"token": {
"display_name": "Token",
"info": "Authentication token for accessing Cassandra on Astra DB.",
"password": True,
},
"database_id": {
"display_name": "Database ID",
"info": "The Astra database ID.",
},
"table_name": {
"display_name": "Table Name",
"info": "The name of the table where vectors will be stored.",
},
"keyspace": {
"display_name": "Keyspace",
"info": "Optional key space within Astra DB. The keyspace should already be created.",
"advanced": True,
},
"ttl_seconds": {
"display_name": "TTL Seconds",
"info": "Optional time-to-live for the added texts.",
"advanced": True,
},
"batch_size": {
"display_name": "Batch Size",
"info": "Optional number of records to process in a single batch.",
"advanced": True,
},
"body_index_options": {
"display_name": "Body Index Options",
"info": "Optional options used to create the body index.",
"advanced": True,
},
"setup_mode": {
"display_name": "Setup Mode",
"info": "Configuration mode for setting up the Cassandra table, with options like 'Sync', 'Async', or 'Off'.",
"options": ["Sync", "Async", "Off"],
"advanced": True,
},
}
def build(
self,
embedding: Embeddings,
token: str,
database_id: str,
inputs: Optional[List[Record]] = None,
keyspace: Optional[str] = None,
table_name: str = "",
ttl_seconds: Optional[int] = None,
batch_size: int = 16,
body_index_options: Optional[List[Tuple[str, Any]]] = None,
setup_mode: SetupMode = SetupMode.SYNC,
) -> VectorStore:
try:
import cassio
except ImportError:
raise ImportError(
"Could not import cassio integration package. " "Please install it with `pip install cassio`."
)
cassio.init(
database_id=database_id,
token=token,
)
if inputs:
documents = [_input.to_lc_document() for _input in inputs]
table = Cassandra.from_documents(
documents=documents,
embedding=embedding,
table_name=table_name,
keyspace=keyspace,
ttl_seconds=ttl_seconds,
batch_size=batch_size,
body_index_options=body_index_options,
)
else:
table = Cassandra(
embedding=embedding,
table_name=table_name,
keyspace=keyspace,
ttl_seconds=ttl_seconds,
body_index_options=body_index_options,
setup_mode=setup_mode,
)
return table

17
src/backend/base/langflow/services/monitor/service.py
View file

@ -1,6 +1,6 @@
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Optional, Union
from typing import TYPE_CHECKING, Optional, Union, List
import duckdb
from loguru import logger
@ -107,12 +107,21 @@ class MonitorService(Service):
return self.exec_query(query)
def delete_messages(self, message_ids: list[int]):
query = f"DELETE FROM messages WHERE index IN ({','.join(map(str, message_ids))})"
def delete_messages(self, message_ids: Union[List[int], str]):
if isinstance(message_ids, list):
# If message_ids is a list, join the string representations of the integers
ids_str = ",".join(map(str, message_ids))
elif isinstance(message_ids, str):
# If message_ids is already a string, use it directly
ids_str = message_ids
else:
raise ValueError("message_ids must be a list of integers or a string")
query = f"DELETE FROM messages WHERE index IN ({ids_str})"
return self.exec_query(query)
def update_message(self, message_id: int, **kwargs):
def update_message(self, message_id: str, **kwargs):
query = (
f"""UPDATE messages SET {', '.join(f"{k} = '{v}'" for k, v in kwargs.items())} WHERE index = {message_id}"""
)

26
src/backend/base/poetry.lock generated
View file

@ -517,13 +517,13 @@ test-randomorder = ["pytest-randomly"]
[[package]]
name = "dataclasses-json"
version = "0.6.6"
version = "0.6.7"
description = "Easily serialize dataclasses to and from JSON."
optional = false
python-versions = "<4.0,>=3.7"
files = [
{file = "dataclasses_json-0.6.6-py3-none-any.whl", hash = "sha256:e54c5c87497741ad454070ba0ed411523d46beb5da102e221efb873801b0ba85"},
{file = "dataclasses_json-0.6.6.tar.gz", hash = "sha256:0c09827d26fffda27f1be2fed7a7a01a29c5ddcd2eb6393ad5ebf9d77e9deae8"},
{file = "dataclasses_json-0.6.7-py3-none-any.whl", hash = "sha256:0dbf33f26c8d5305befd61b39d2b3414e8a407bedc2834dea9b8d642666fb40a"},
{file = "dataclasses_json-0.6.7.tar.gz", hash = "sha256:b6b3e528266ea45b9535223bc53ca645f5208833c29229e847b3f26a1cc55fc0"},
]
[package.dependencies]
@ -1148,13 +1148,13 @@ jsonpointer = ">=1.9"
[[package]]
name = "jsonpointer"
version = "2.4"
version = "3.0.0"
description = "Identify specific nodes in a JSON document (RFC 6901)"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, !=3.5.*, !=3.6.*"
python-versions = ">=3.7"
files = [
{file = "jsonpointer-2.4-py2.py3-none-any.whl", hash = "sha256:15d51bba20eea3165644553647711d150376234112651b4f1811022aecad7d7a"},
{file = "jsonpointer-2.4.tar.gz", hash = "sha256:585cee82b70211fa9e6043b7bb89db6e1aa49524340dde8ad6b63206ea689d88"},
{file = "jsonpointer-3.0.0-py2.py3-none-any.whl", hash = "sha256:13e088adc14fca8b6aa8177c044e12701e6ad4b28ff10e65f2267a90109c9942"},
{file = "jsonpointer-3.0.0.tar.gz", hash = "sha256:2b2d729f2091522d61c3b31f82e11870f60b68f43fbc705cb76bf4b832af59ef"},
]
[[package]]
@ -1260,13 +1260,13 @@ extended-testing = ["beautifulsoup4 (>=4.12.3,<5.0.0)", "lxml (>=4.9.3,<6.0)"]
[[package]]
name = "langchainhub"
version = "0.1.17"
version = "0.1.18"
description = "The LangChain Hub API client"
optional = false
python-versions = "<4.0,>=3.8.1"
files = [
{file = "langchainhub-0.1.17-py3-none-any.whl", hash = "sha256:4c609b3948252c71670f0d98f73413b515cfd2f6701a7b40ce959203e6133e04"},
{file = "langchainhub-0.1.17.tar.gz", hash = "sha256:af7df0cb1cebc7a6e0864e8632ae48ecad39ed96568f699c78657b9d04e50b46"},
{file = "langchainhub-0.1.18-py3-none-any.whl", hash = "sha256:11501f15e7f34715ecc8892587daa35c6f2a3005e1f2926c9bcabd31fc2c100c"},
{file = "langchainhub-0.1.18.tar.gz", hash = "sha256:f2d0d8bf3abe4ca5e70511d8220bdc9ccea28d5267bcfd0e5ef9c53bd5bd3bad"},
]
[package.dependencies]
@ -2713,13 +2713,13 @@ urllib3 = ">=2"
[[package]]
name = "typing-extensions"
version = "4.12.1"
version = "4.12.2"
description = "Backported and Experimental Type Hints for Python 3.8+"
optional = false
python-versions = ">=3.8"
files = [
{file = "typing_extensions-4.12.1-py3-none-any.whl", hash = "sha256:6024b58b69089e5a89c347397254e35f1bf02a907728ec7fee9bf0fe837d203a"},
{file = "typing_extensions-4.12.1.tar.gz", hash = "sha256:915f5e35ff76f56588223f15fdd5938f9a1cf9195c0de25130c627e4d597f6d1"},
{file = "typing_extensions-4.12.2-py3-none-any.whl", hash = "sha256:04e5ca0351e0f3f85c6853954072df659d0d13fac324d0072316b67d7794700d"},
{file = "typing_extensions-4.12.2.tar.gz", hash = "sha256:1a7ead55c7e559dd4dee8856e3a88b41225abfe1ce8df57b7c13915fe121ffb8"},
]
[[package]]