jyapayne/langflow
Watch
1
0
Fork
You've already forked langflow
 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Merge branch 'dev' into fixFlowSettings

Browse source
This commit is contained in:
anovazzi1 2024-06-26 15:38:04 -03:00 • committed by GitHub
commit c6afb0f999
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
64 changed files with 993 additions and 436 deletions
Download patch file Download diff file Expand all files Collapse all files

14
.github/workflows/auto-update.yml vendored Normal file
View file

@ -0,0 +1,14 @@
name: Auto-update
on:
push:
branches:
- dev
- main
jobs:
Auto:
name: Auto-update
runs-on: ubuntu-latest
steps:
- uses: tibdex/auto-update@v2

2
.github/workflows/lint-js.yml vendored
View file

@ -5,7 +5,7 @@ on:
paths:
- "src/frontend/**"
merge_group:
branches: [dev]
types: [checks_requested]
env:
NODE_VERSION: "21"

2
.github/workflows/python_test.yml vendored
View file

@ -45,7 +45,7 @@ jobs:
poetry run python -m langflow run --host 127.0.0.1 --port 7860 --backend-only &
SERVER_PID=$!
# Wait for the server to start
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/auto_login; do sleep 2; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/api/v1/auto_login; do sleep 5; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
# Terminate the server
kill $SERVER_PID || (echo "Failed to terminate the server" && exit 1)
sleep 10 # give the server some time to terminate

4
.github/workflows/release.yml vendored
View file

@ -62,7 +62,7 @@ jobs:
python -m langflow run --host 127.0.0.1 --port 7860 &
SERVER_PID=$!
# Wait for the server to start
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/auto_login; do sleep 2; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/api/v1/auto_login; do sleep 2; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
# Terminate the server
kill $SERVER_PID || (echo "Failed to terminate the server" && exit 1)
sleep 10 # give the server some time to terminate
@ -124,7 +124,7 @@ jobs:
python -m langflow run --host 127.0.0.1 --port 7860 &
SERVER_PID=$!
# Wait for the server to start
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/auto_login; do sleep 2; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
timeout 120 bash -c 'until curl -f http://127.0.0.1:7860/api/v1/auto_login; do sleep 2; done' || (echo "Server did not start in time" && kill $SERVER_PID && exit 1)
# Terminate the server
kill $SERVER_PID || (echo "Failed to terminate the server" && exit 1)
sleep 10 # give the server some time to terminate

53
Makefile
View file

@ -47,20 +47,22 @@ init:
coverage: ## run the tests and generate a coverage report
poetry run pytest --cov \
--cov-config=.coveragerc \
--cov-report xml \
--cov-report term-missing:skip-covered \
--cov-report lcov:coverage/lcov-pytest.info
@poetry run coverage run
@poetry run coverage erase
# allow passing arguments to pytest
unit_tests:
poetry run pytest --ignore=tests/integration --instafail -ra -n auto -m "not api_key_required" $(args)
poetry run pytest \
--ignore=tests/integration \
--instafail -ra -n auto -m "not api_key_required" \
$(args)
integration_tests:
poetry run pytest tests/integration --instafail -ra -n auto $(args)
poetry run pytest tests/integration \
--instafail -ra -n auto \
$(args)
format: ## run code formatters
poetry run ruff check . --fix
@ -129,9 +131,20 @@ start:
@echo 'Running the CLI'
ifeq ($(open_browser),false)
@make install_backend && poetry run langflow run --path $(path) --log-level $(log_level) --host $(host) --port $(port) --env-file $(env) --no-open-browser
@make install_backend && poetry run langflow run \
--path $(path) \
--log-level $(log_level) \
--host $(host) \
--port $(port) \
--env-file $(env) \
--no-open-browser
else
@make install_backend && poetry run langflow run --path $(path) --log-level $(log_level) --host $(host) --port $(port) --env-file $(env)
@make install_backend && poetry run langflow run \
--path $(path) \
--log-level $(log_level) \
--host $(host) \
--port $(port) \
--env-file $(env)
endif
@ -166,13 +179,27 @@ backend: ## run the backend in development mode
@echo 'Setting up the environment'
@make setup_env
make install_backend
@-kill -9 $(lsof -t -i:7860)
@-kill -9 $$(lsof -t -i:7860)
ifdef login
@echo "Running backend autologin is $(login)";
LANGFLOW_AUTO_LOGIN=$(login) poetry run uvicorn --factory langflow.main:create_app --host 0.0.0.0 --port 7860 --reload --env-file .env --loop asyncio --workers $(workers)
LANGFLOW_AUTO_LOGIN=$(login) poetry run uvicorn \
--factory langflow.main:create_app \
--host 0.0.0.0 \
--port $(port) \
--reload \
--env-file $(env) \
--loop asyncio \
--workers $(workers)
else
@echo "Running backend respecting the .env file";
poetry run uvicorn --factory langflow.main:create_app --host 0.0.0.0 --port 7860 --reload --env-file .env --loop asyncio --workers $(workers)
@echo "Running backend respecting the $(env) file";
poetry run uvicorn \
--factory langflow.main:create_app \
--host 0.0.0.0 \
--port $(port) \
--reload \
--env-file $(env) \
--loop asyncio \
--workers $(workers)
endif

317
docs/docs/deployment/kubernetes.md Normal file
View file

@ -0,0 +1,317 @@
# Kubernetes
This guide will help you get LangFlow up and running in Kubernetes cluster, including the following steps:
- Install [LangFlow as IDE](#langflow-ide) in a Kubernetes cluster (for development)
- Install [LangFlow as a standalone application](#langflow-runtime) in a Kubernetes cluster (for production runtime workloads)
## LangFlow (IDE)
This solution is designed to provide a complete environment for developers to create, test, and debug their flows. It includes both the API and the UI.
### Prerequisites
- Kubernetes server
- kubectl
- Helm
### Step 0. Prepare a Kubernetes cluster
We use [Minikube](https://minikube.sigs.k8s.io/docs/start/) for this example, but you can use any Kubernetes cluster.
1. Create a Kubernetes cluster on Minikube.
```shell
minikube start
```
2. Set `kubectl` to use Minikube.
```shell
kubectl config use-context minikube
```
### Step 1. Install the LangFlow Helm chart
1. Add the repository to Helm.
```shell
helm repo add langflow https://langflow-ai.github.io/langflow-helm-charts
helm repo update
```
2. Install LangFlow with the default options in the `langflow` namespace.
```shell
helm install langflow-ide langflow/langflow-ide -n langflow --create-namespace
```
3. Check the status of the pods
```shell
kubectl get pods -n langflow
```
```
NAME READY STATUS RESTARTS AGE
langflow-0 1/1 Running 0 33s
langflow-frontend-5d9c558dbb-g7tc9 1/1 Running 0 38s
```
### Step 2. Access LangFlow
Enable local port forwarding to access LangFlow from your local machine.
```shell
kubectl port-forward -n langflow svc/langflow-langflow-runtime 7860:7860
```
Now you can access LangFlow at [http://localhost:7860/](http://localhost:7860/).
### LangFlow version
To specify a different LangFlow version, you can set the `langflow.backend.image.tag` and `langflow.frontend.image.tag` values in the `values.yaml` file.
```yaml
langflow:
backend:
image:
tag: "1.0.0a59"
frontend:
image:
tag: "1.0.0a59"
```
### Storage
By default, the chart will use a SQLLite database stored in a local persistent disk.
If you want to use an external PostgreSQL database, you can set the `langflow.database` values in the `values.yaml` file.
```yaml
# Deploy postgresql. You can skip this section if you have an existing postgresql database.
postgresql:
enabled: true
fullnameOverride: "langflow-ide-postgresql-service"
auth:
username: "langflow"
password: "langflow-postgres"
database: "langflow-db"
langflow:
backend:
externalDatabase:
enabled: true
driver:
value: "postgresql"
host:
value: "langflow-ide-postgresql-service"
port:
value: "5432"
database:
value: "langflow-db"
user:
value: "langflow"
password:
valueFrom:
secretKeyRef:
key: "password"
name: "langflow-ide-postgresql-service"
sqlite:
enabled: false
```
### Scaling
You can scale the number of replicas for the LangFlow backend and frontend services by changing the `replicaCount` value in the `values.yaml` file.
```yaml
langflow:
backend:
replicaCount: 3
frontend:
replicaCount: 3
```
You can scale frontend and backend services independently.
To scale vertically (increase the resources for the pods), you can set the `resources` values in the `values.yaml` file.
```yaml
langflow:
backend:
resources:
requests:
memory: "2Gi"
cpu: "1000m"
frontend:
resources:
requests:
memory: "1Gi"
cpu: "1000m"
```
### Deploy on AWS EKS, Google GKE, or Azure AKS and other examples
Visit the [LangFlow Helm Charts repository](https://github.com/langflow-ai/langflow-helm-charts) for more examples and configurations.
Use the [default values file](https://github.com/langflow-ai/langflow-helm-charts/blob/main/charts/langflow-ide/values.yaml) as reference for all the options available.
Visit the [examples directory](https://github.com/langflow-ai/langflow-helm-charts/tree/main/examples/langflow-ide) to learn more about different deployment options.
## LangFlow (Runtime)
The runtime chart is tailored for deploying applications in a production environment. It is focused on stability, performance, isolation and security to ensure that applications run reliably and efficiently.
Using a dedicated deployment for a set of flows is fundamental in production environments in order to have a granular resource control.
## Import a flow
There are two ways to import a flow (or multiple flows) and deploy it with the LangFlow runtime Helm chart:
1. **From a remote location**: You can reference a flow stored in a remote location, such as a URL or a Git repository by customizing the `values.yaml` file in the section `downloadFlows`:
```yaml
downloadFlows:
flows:
- url: https://raw.githubusercontent.com/langflow-ai/langflow/dev/src/backend/base/langflow/initial_setup/starter_projects/Basic%20Prompting%20(Hello%2C%20world!).json
```
When the LangFlow runtime starts, it will download the flow from the specified URL and run it.
The flow UUID to use to call the API endpoints is the same as the one in the JSON file under the `id` field.
You can also specify a `endpoint_name` field to give a friendly name to the flow.
2. **Packaging the flow as docker image**: You can add a flow from to a docker image based on Langflow runtime and refer to it in the chart.
First you need a base Dockerfile to get the langflow image and add your local flows:
```Dockerfile
FROM langflowai/langflow-backend:latest
RUN mkdir /app/flows
COPY ./*json /app/flows/.
```
Then you can build the image and push it to DockerHub (or any registry you prefer):
```bash
# Create the Dockerfile
echo """FROM langflowai/langflow-backend:latest
RUN mkdir /app/flows
ENV LANGFLOW_LOAD_FLOWS_PATH=/app/flows
COPY ./*json /app/flows/.""" > Dockerfile
# Download the flows
wget https://raw.githubusercontent.com/langflow-ai/langflow/dev/src/backend/base/langflow/initial_setup/starter_projects/Basic%20Prompting%20(Hello%2C%20world!).json
# Build the docker image locally
docker build -t myuser/langflow-just-chat:1.0.0 -f Dockerfile .
# Push the image to DockerHub
docker push myuser/langflow-just-chat:1.0.0
```
### Prerequisites
- Kubernetes server
- kubectl
- Helm
### Step 0. Prepare a Kubernetes cluster
Follow the same steps as for the LangFlow IDE.
### Step 1. Install the LangFlow runtime Helm chart
1. Add the repository to Helm.
```shell
helm repo add langflow https://langflow-ai.github.io/langflow-helm-charts
helm repo update
```
2. Install the LangFlow app with the default options in the `langflow` namespace.
If you bundled the flow in a docker image, you can specify the image name in the `values.yaml` file or with the `--set` flag:
```shell
helm install my-langflow-app langflow/langflow-runtime -n langflow --create-namespace --set image.repository=myuser/langflow-just-chat --set image.tag=1.0.0
```
If you want to download the flow from a remote location, you can specify the URL in the `values.yaml` file or with the `--set` flag:
```shell
helm install my-langflow-app langflow/langflow-runtime -n langflow --create-namespace --set downloadFlows.flows[0].url=https://raw.githubusercontent.com/langflow-ai/langflow/dev/src/backend/base/langflow/initial_setup/starter_projects/Basic%20Prompting%20(Hello%2C%20world!).json
```
3. Check the status of the pods
```shell
kubectl get pods -n langflow
```
### Step 2. Access the LangFlow app API
Enable local port forwarding to access LangFlow from your local machine.
```shell
kubectl port-forward -n langflow svc/langflow-my-langflow-app 7860:7860
```
Now you can access the API at [http://localhost:7860/api/v1/flows](http://localhost:7860/api/v1/flows) and execute the flow:
```bash
id=$(curl -s http://localhost:7860/api/v1/flows | jq -r '.flows[0].id')
curl -X POST \
"http://localhost:7860/api/v1/run/$id?stream=false" \
-H 'Content-Type: application/json'\
-d '{
"input_value": "Hello!",
"output_type": "chat",
"input_type": "chat"
}'
```
### Storage
In this case, the storage is not needed as our deployment is stateless.
### Log level and LangFlow configurations
You can set the log level and other LangFlow configurations in the `values.yaml` file.
```yaml
env:
- name: LANGFLOW_LOG_LEVEL
value: "INFO"
```
### Configure secrets and variables
In order to inject secrets and LangFlow global variables, you can use the `secrets` and `env` sections in the `values.yaml` file.
Let's say your flow uses a global variable which is a secret; when you export the flow as JSON, it's recommended to not include it.
When importing the flow in the LangFlow runtime, you can set the global variable using the `env` section in the `values.yaml` file.
Assuming you have a global variable called `openai_key_var`, you can read it directly from a secret:
```yaml
env:
- name: openai_key_var
valueFrom:
secretKeyRef:
name: openai-key
key: openai-key
```
or directly from the values file (not recommended for secret values!):
```yaml
env:
- name: openai_key_var
value: "sk-...."
```
### Scaling
You can scale the number of replicas for the LangFlow app by changing the `replicaCount` value in the `values.yaml` file.
```yaml
replicaCount: 3
```
To scale vertically (increase the resources for the pods), you can set the `resources` values in the `values.yaml` file.
```yaml
resources:
requests:
memory: "2Gi"
cpu: "1000m"
```
### Other examples
Visit the [LangFlow Helm Charts repository](https://github.com/langflow-ai/langflow-helm-charts) for more examples and configurations.
Use the [default values file](https://github.com/langflow-ai/langflow-helm-charts/blob/main/charts/langflow-runtime/values.yaml) as reference for all the options available.
Visit the [examples directory](https://github.com/langflow-ai/langflow-helm-charts/tree/main/examples/langflow-runtime) to learn more about different deployment options.

22
pyproject.toml
View file

@ -151,6 +151,28 @@ log_cli = true
markers = ["async_test", "api_key_required"]
[tool.coverage.run]
command_line = """
-m pytest
--cov --cov-report=term --cov-report=html
--instafail -ra -n auto -m "not api_key_required"
tests/unit
"""
source = ["src/backend/base/langflow/"]
omit = ["*/alembic/*", "tests/*", "*/__init__.py"]
[tool.coverage.report]
sort = "Stmts"
skip_empty = true
show_missing = false
ignore_errors = true
[tool.coverage.html]
directory = "coverage"
[tool.ruff]
exclude = ["src/backend/langflow/alembic/*"]
line-length = 120

52
src/backend/base/langflow/alembic/versions/d066bfd22890_add_message_table.py Normal file
View file

@ -0,0 +1,52 @@
"""Add message table
Revision ID: 325180f0c4e1
Revises: 631faacf5da2
Create Date: 2024-06-23 21:29:28.220100
"""
from typing import Sequence, Union
import sqlalchemy as sa
import sqlmodel
from alembic import op
from langflow.utils import migration
# revision identifiers, used by Alembic.
revision: str = "325180f0c4e1"
down_revision: Union[str, None] = "631faacf5da2"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
conn = op.get_bind()
# ### commands auto generated by Alembic - please adjust! ###
if not migration.table_exists("message", conn):
op.create_table(
"message",
sa.Column("timestamp", sa.DateTime(), nullable=False),
sa.Column("sender", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("sender_name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("session_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("text", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
sa.Column("flow_id", sqlmodel.sql.sqltypes.GUID(), nullable=True),
sa.Column("files", sa.JSON(), nullable=True),
sa.ForeignKeyConstraint(
["flow_id"],
["flow.id"],
),
sa.PrimaryKeyConstraint("id"),
)
# ### end Alembic commands ###
def downgrade() -> None:
conn = op.get_bind()
# ### commands auto generated by Alembic - please adjust! ###
if migration.table_exists("message", conn):
op.drop_table("message")
# ### end Alembic commands ###

84
src/backend/base/langflow/api/v1/monitor.py
View file

@ -1,15 +1,15 @@
from typing import List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import delete
from sqlmodel import Session, col, select
from langflow.services.deps import get_monitor_service
from langflow.services.monitor.schema import (
MessageModelRequest,
MessageModelResponse,
TransactionModelResponse,
VertexBuildMapModel,
)
from langflow.services.auth.utils import get_current_active_user
from langflow.services.database.models.message.model import MessageRead, MessageTable, MessageUpdate
from langflow.services.database.models.user.model import User
from langflow.services.deps import get_monitor_service, get_session
from langflow.services.monitor.schema import MessageModelResponse, TransactionModelResponse, VertexBuildMapModel
from langflow.services.monitor.service import MonitorService
router = APIRouter(prefix="/monitor", tags=["Monitor"])
@ -52,45 +52,58 @@ async def get_messages(
sender: Optional[str] = Query(None),
sender_name: Optional[str] = Query(None),
order_by: Optional[str] = Query("timestamp"),
monitor_service: MonitorService = Depends(get_monitor_service),
session: Session = Depends(get_session),
):
try:
df = monitor_service.get_messages(
flow_id=flow_id,
sender=sender,
sender_name=sender_name,
session_id=session_id,
order_by=order_by,
)
dicts = df.to_dict(orient="records")
return [MessageModelResponse(**d) for d in dicts]
stmt = select(MessageTable)
if flow_id:
stmt = stmt.where(MessageTable.flow_id == flow_id)
if session_id:
stmt = stmt.where(MessageTable.session_id == session_id)
if sender:
stmt = stmt.where(MessageTable.sender == sender)
if sender_name:
stmt = stmt.where(MessageTable.sender_name == sender_name)
if order_by:
col = getattr(MessageTable, order_by).asc()
stmt = stmt.order_by(col)
messages = session.exec(stmt)
return [MessageModelResponse.model_validate(d, from_attributes=True) for d in messages]
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/messages", status_code=204)
async def delete_messages(
message_ids: List[int],
monitor_service: MonitorService = Depends(get_monitor_service),
message_ids: List[UUID],
session: Session = Depends(get_session),
current_user: User = Depends(get_current_active_user),
):
try:
monitor_service.delete_messages(message_ids=message_ids)
session.exec(select(MessageTable).where(MessageTable.id.in_(message_ids))) # type: ignore
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/messages/{message_id}", response_model=MessageModelResponse)
@router.put("/messages/{message_id}", response_model=MessageRead)
async def update_message(
message_id: int,
message: MessageModelRequest,
monitor_service: MonitorService = Depends(get_monitor_service),
message_id: UUID,
message: MessageUpdate,
session: Session = Depends(get_session),
user: User = Depends(get_current_active_user),
):
try:
message_dict = message.model_dump(exclude_none=True)
message_dict.pop("index", None)
monitor_service.update_message(message_id=message_id, **message_dict) # type: ignore
return MessageModelResponse(index=message_id, **message_dict)
db_message = session.get(MessageTable, message_id)
if not db_message:
raise HTTPException(status_code=404, detail="Message not found")
message_dict = message.model_dump(exclude_unset=True, exclude_none=True)
db_message.sqlmodel_update(message_dict)
session.add(db_message)
session.commit()
session.refresh(db_message)
return db_message
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@ -98,10 +111,16 @@ async def update_message(
@router.delete("/messages/session/{session_id}", status_code=204)
async def delete_messages_session(
session_id: str,
monitor_service: MonitorService = Depends(get_monitor_service),
session: Session = Depends(get_session),
):
try:
monitor_service.delete_messages_session(session_id=session_id)
session.exec( # type: ignore
delete(MessageTable)
.where(col(MessageTable.session_id) == session_id)
.execution_options(synchronize_session="fetch")
)
session.commit()
return {"message": "Messages deleted successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@ -137,4 +156,3 @@ async def get_transactions(
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
raise HTTPException(status_code=500, detail=str(e))

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

@ -119,7 +119,11 @@ class LCModelComponent(Component):
return status_message
def get_chat_result(
self, runnable: LanguageModel, stream: bool, input_value: str | Message, system_message: Optional[str] = None
self,
runnable: LanguageModel,
stream: bool,
input_value: str | Message,
system_message: Optional[str] = None,
):
messages: list[Union[BaseMessage]] = []
if not input_value and not system_message:

2
src/backend/base/langflow/custom/custom_component/custom_component.py
View file

@ -96,7 +96,7 @@ class CustomComponent(BaseComponent):
def stop(self, output_name: str | None = None):
if not output_name and self.vertex and len(self.vertex.outputs) == 1:
output_name = self.vertex.outputs[0]["name"]
else:
elif not output_name:
raise ValueError("You must specify an output name to call stop")
if not self.vertex:
raise ValueError("Vertex is not set")

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

@ -1208,6 +1208,7 @@ class Graph:
except ValueError:
stop_or_start_vertex = self.get_root_of_group_node(vertex_id)
stack = [stop_or_start_vertex.id]
vertex_id = stop_or_start_vertex.id
stop_predecessors = [pre.id for pre in stop_or_start_vertex.predecessors]
# DFS to collect all vertices that can reach the specified vertex
while stack:

97
src/backend/base/langflow/memory.py
View file

@ -1,11 +1,14 @@
import warnings
from typing import List, Optional
from uuid import UUID
from loguru import logger
from sqlalchemy import delete
from sqlmodel import Session, col, select
from langflow.schema.message import Message
from langflow.services.deps import get_monitor_service
from langflow.services.monitor.schema import MessageModel
from langflow.services.database.models.message.model import MessageRead, MessageTable
from langflow.services.deps import session_scope
def get_messages(
@ -14,6 +17,7 @@ def get_messages(
session_id: Optional[str] = None,
order_by: Optional[str] = "timestamp",
order: Optional[str] = "DESC",
flow_id: Optional[UUID] = None,
limit: Optional[int] = None,
):
"""
@ -29,34 +33,29 @@ def get_messages(
Returns:
List[Data]: A list of Data objects representing the retrieved messages.
"""
monitor_service = get_monitor_service()
messages_df = monitor_service.get_messages(
sender=sender,
sender_name=sender_name,
session_id=session_id,
order_by=order_by,
limit=limit,
order=order,
)
messages_read: list[Message] = []
with session_scope() as session:
stmt = select(MessageTable)
if sender:
stmt = stmt.where(MessageTable.sender == sender)
if sender_name:
stmt = stmt.where(MessageTable.sender_name == sender_name)
if session_id:
stmt = stmt.where(MessageTable.session_id == session_id)
if flow_id:
stmt = stmt.where(MessageTable.flow_id == flow_id)
if order_by:
if order == "DESC":
col = getattr(MessageTable, order_by).desc()
else:
col = getattr(MessageTable, order_by).asc()
stmt = stmt.order_by(col)
if limit:
stmt = stmt.limit(limit)
messages = session.exec(stmt)
messages_read = [Message(**d.model_dump()) for d in messages]
messages: list[Message] = []
# messages_df has a timestamp
# it gets the last 5 messages, for example
# but now they are ordered from most recent to least recent
# so we need to reverse the order
messages_df = messages_df[::-1] if order == "DESC" else messages_df
for row in messages_df.itertuples():
msg = Message(
text=row.text,
sender=row.sender,
session_id=row.session_id,
sender_name=row.sender_name,
timestamp=row.timestamp,
)
messages.append(msg)
return messages
return messages_read
def add_messages(messages: Message | list[Message], flow_id: Optional[str] = None):
@ -64,7 +63,6 @@ def add_messages(messages: Message | list[Message], flow_id: Optional[str] = Non
Add a message to the monitor service.
"""
try:
monitor_service = get_monitor_service()
if not isinstance(messages, list):
messages = [messages]
@ -72,25 +70,29 @@ def add_messages(messages: Message | list[Message], flow_id: Optional[str] = Non
types = ", ".join([str(type(message)) for message in messages])
raise ValueError(f"The messages must be instances of Message. Found: {types}")
messages_models: list[MessageModel] = []
messages_models: list[MessageTable] = []
for msg in messages:
if not msg.timestamp:
msg.timestamp = monitor_service.get_timestamp()
messages_models.append(MessageModel.from_message(msg, flow_id=flow_id))
for message_model in messages_models:
try:
monitor_service.add_message(message_model)
except Exception as e:
logger.error(f"Error adding message to monitor service: {e}")
logger.exception(e)
raise e
return messages_models
messages_models.append(MessageTable.from_message(msg, flow_id=flow_id))
with session_scope() as session:
messages_models = add_messagetables(messages_models, session)
return [Message(**message.model_dump()) for message in messages_models]
except Exception as e:
logger.exception(e)
raise e
def add_messagetables(messages: list[MessageTable], session: Session):
for message in messages:
try:
session.add(message)
session.commit()
session.refresh(message)
except Exception as e:
logger.exception(e)
raise e
return [MessageRead.model_validate(message, from_attributes=True) for message in messages]
def delete_messages(session_id: str):
"""
Delete messages from the monitor service based on the provided session ID.
@ -98,8 +100,13 @@ def delete_messages(session_id: str):
Args:
session_id (str): The session ID associated with the messages to delete.
"""
monitor_service = get_monitor_service()
monitor_service.delete_messages_session(session_id)
with session_scope() as session:
session.exec(
delete(MessageTable)
.where(col(MessageTable.session_id) == session_id)
.execution_options(synchronize_session="fetch")
)
session.commit()
def store_message(

10
src/backend/base/langflow/schema/message.py
View file

@ -1,5 +1,6 @@
from datetime import datetime, timezone
from typing import Annotated, Any, AsyncIterator, Iterator, List, Optional
from uuid import UUID
from fastapi.encoders import jsonable_encoder
from langchain_core.load import load
@ -31,7 +32,14 @@ class Message(Data):
timestamp: Annotated[str, BeforeValidator(_timestamp_to_str)] = Field(
default=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
)
flow_id: Optional[str] = None
flow_id: Optional[str | UUID] = None
@field_validator("flow_id", mode="before")
@classmethod
def validate_flow_id(cls, value):
if isinstance(value, UUID):
value = str(value)
return value
@field_validator("files", mode="before")
@classmethod

3
src/backend/base/langflow/services/database/models/__init__.py
View file

@ -1,7 +1,8 @@
from .api_key import ApiKey
from .flow import Flow
from .folder import Folder
from .message import MessageTable
from .user import User
from .variable import Variable
__all__ = ["Flow", "User", "ApiKey", "Variable", "Folder"]
__all__ = ["Flow", "User", "ApiKey", "Variable", "Folder", "MessageTable"]

4
src/backend/base/langflow/services/database/models/flow/model.py
View file

@ -3,7 +3,7 @@
import re
import warnings
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Dict, Optional
from typing import TYPE_CHECKING, Dict, List, Optional
from uuid import UUID, uuid4
import emoji
@ -17,6 +17,7 @@ from langflow.schema import Data
if TYPE_CHECKING:
from langflow.services.database.models.folder import Folder
from langflow.services.database.models.message import MessageTable
from langflow.services.database.models.user import User
@ -141,6 +142,7 @@ class Flow(FlowBase, table=True):
user: "User" = Relationship(back_populates="flows")
folder_id: Optional[UUID] = Field(default=None, foreign_key="folder.id", nullable=True, index=True)
folder: Optional["Folder"] = Relationship(back_populates="flows")
messages: List["MessageTable"] = Relationship(back_populates="flow")
def to_data(self):
serialized = self.model_dump()

3
src/backend/base/langflow/services/database/models/message/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .model import MessageTable, MessageCreate, MessageRead, MessageUpdate
__all__ = ["MessageTable", "MessageCreate", "MessageRead", "MessageUpdate"]

74
src/backend/base/langflow/services/database/models/message/model.py Normal file
View file

@ -0,0 +1,74 @@
from datetime import datetime, timezone
from typing import TYPE_CHECKING, List, Optional
from uuid import UUID, uuid4
from pydantic import field_validator
from sqlmodel import JSON, Column, Field, Relationship, SQLModel
if TYPE_CHECKING:
from langflow.schema.message import Message
from langflow.services.database.models.flow.model import Flow
class MessageBase(SQLModel):
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
sender: str
sender_name: str
session_id: str
text: str
files: list[str] = Field(default_factory=list)
@field_validator("files", mode="before")
@classmethod
def validate_files(cls, value):
if not value:
value = []
return value
@classmethod
def from_message(cls, message: "Message", flow_id: str | None = None):
# first check if the record has all the required fields
if message.text is None or not message.sender or not message.sender_name:
raise ValueError("The message does not have the required fields (text, sender, sender_name).")
if isinstance(message.timestamp, str):
timestamp = datetime.fromisoformat(message.timestamp)
else:
timestamp = message.timestamp
return cls(
sender=message.sender,
sender_name=message.sender_name,
text=message.text,
session_id=message.session_id,
files=message.files or [],
timestamp=timestamp,
flow_id=flow_id,
)
class MessageTable(MessageBase, table=True):
__tablename__ = "message"
id: UUID = Field(default_factory=uuid4, primary_key=True)
flow_id: Optional[UUID] = Field(default=None, foreign_key="flow.id")
flow: "Flow" = Relationship(back_populates="messages")
files: List[str] = Field(sa_column=Column(JSON))
# Needed for Column(JSON)
class Config:
arbitrary_types_allowed = True
class MessageRead(MessageBase):
id: UUID
flow_id: Optional[UUID] = Field()
class MessageCreate(MessageBase):
pass
class MessageUpdate(SQLModel):
text: Optional[str] = None
sender: Optional[str] = None
sender_name: Optional[str] = None
session_id: Optional[str] = None
files: Optional[list[str]] = None

16
src/backend/base/langflow/services/monitor/schema.py
View file

@ -1,6 +1,7 @@
import json
from datetime import datetime, timezone
from typing import Any, Optional
from uuid import UUID
from pydantic import BaseModel, Field, field_serializer, field_validator
@ -81,8 +82,8 @@ class TransactionModelResponse(DefaultModel):
class MessageModel(DefaultModel):
index: Optional[int] = Field(default=None)
flow_id: Optional[str] = Field(default=None, alias="flow_id")
id: Optional[str | UUID] = Field(default=None)
flow_id: Optional[UUID] = Field(default=None)
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
sender: str
sender_name: str
@ -127,16 +128,7 @@ class MessageModel(DefaultModel):
class MessageModelResponse(MessageModel):
index: Optional[int] = Field(default=None)
@field_validator("index", mode="before")
def validate_id(cls, v):
if isinstance(v, float):
try:
return int(v)
except ValueError:
return None
return v
pass
class MessageModelRequest(MessageModel):

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

@ -3,14 +3,15 @@ from pathlib import Path
from typing import TYPE_CHECKING, List, Optional, Union
import duckdb
from langflow.services.base import Service
from langflow.services.monitor.utils import add_row_to_table, drop_and_create_table_if_schema_mismatch
from loguru import logger
from platformdirs import user_cache_dir
from langflow.services.base import Service
from langflow.services.monitor.utils import add_row_to_table, drop_and_create_table_if_schema_mismatch
if TYPE_CHECKING:
from langflow.services.settings.service import SettingsService
from langflow.services.monitor.schema import MessageModel, TransactionModel, VertexBuildModel
from langflow.services.settings.service import SettingsService
class MonitorService(Service):
@ -129,45 +130,6 @@ class MonitorService(Service):
return self.exec_query(query, read_only=False)
def add_message(self, message: "MessageModel"):
self.add_row("messages", message)
def get_messages(
self,
flow_id: Optional[str] = None,
sender: Optional[str] = None,
sender_name: Optional[str] = None,
session_id: Optional[str] = None,
order_by: Optional[str] = "timestamp",
order: Optional[str] = "DESC",
limit: Optional[int] = None,
):
query = "SELECT index, flow_id, sender_name, sender, session_id, text, files, timestamp FROM messages"
conditions = []
if sender:
conditions.append(f"sender = '{sender}'")
if sender_name:
conditions.append(f"sender_name = '{sender_name}'")
if session_id:
conditions.append(f"session_id = '{session_id}'")
if flow_id:
conditions.append(f"flow_id = '{flow_id}'")
if conditions:
query += " WHERE " + " AND ".join(conditions)
if order_by and order:
# Make sure the order is from newest to oldest
query += f" ORDER BY {order_by} {order.upper()}"
if limit is not None:
query += f" LIMIT {limit}"
with duckdb.connect(str(self.db_path), read_only=True) as conn:
df = conn.execute(query).df()
return df
def get_transactions(
self,
source: Optional[str] = None,

Some files were not shown because too many files have changed in this diff Show more