Flow building each time when use langflow as a python code

[Preet1234-ux]

I am trying to use my flow build in langflow as a python code as below :

from langflow import load_flow_from_json
TWEAKS = {
  "LLMCheckerChain-0xJIZ": {},
  "LlamaComponent-14J2Z": {}
}
flow = load_flow_from_json("Untitled document.json", tweaks=TWEAKS)
# Now you can use it like any chain
inputs = {"query":""}
flow(inputs)

But each time when I am entering a new query , everytime build is building which is taking time for each new query and I dont want that . I want build to happen only once and then for every query full process should not repeat .

Please suggest a way to do that ?

[dosubot[bot]]

To address the issue of the flow being rebuilt with each new query in your langflow application, you can optimize the process by implementing caching mechanisms. This approach involves checking if the flow already exists in the cache before deciding to rebuild it, thus saving time by reusing the built flow for subsequent queries. Here's a streamlined way to achieve this:

1. Implement Caching: Use caching utilities to check for an existing flow in the cache. If the flow is not found, build it and then save the built flow to the cache. This ensures the flow is built only once.

2. Utilize load_flow_from_json and run_flow_from_json Functions: These functions from src/backend/base/langflow/load.py are designed for efficient loading and running of your flow. Use load_flow_from_json to load the flow from the cache and run_flow_from_json to apply it to new queries without rebuilding.

3. Adapt Caching Utilities: Adapt functions from src/backend/base/langflow/services/cache/utils.py for managing the cache, including creating a cache directory, saving the flow, and checking for its existence before building.

4. Example Code Snippet:

   # Pseudocode for checking cache and loading flow
   if not os.path.exists(flow_path):
       flow = build_flow()  # Your function to build the flow
       save_flow(flow)  # Save the built flow for future use
   else:
       flow = load_flow_from_json(flow_path)  # Load the existing flow

By following these steps, you ensure that the flow is built only once and reused for each new query, significantly reducing the processing time and improving the efficiency of your application.

Details

Sources

• src/backend/base/langflow/services/cache/service.py
• src/backend/base/langflow/services/cache/utils.py
• src/backend/base/langflow/load.py
• src/backend/base/langflow/processing/load.py
• src/backend/langflow/processing/load.py
• src/frontend/tests/end-to-end/assets/flow.json
• src/frontend/tests/onlyFront/assets/flow.json
• src/backend/langflow/services/cache/service.py
• src/backend/base/langflow/helpers/flow.py
• src/frontend/src/stores/flowStore.ts
• src/frontend/src/stores/flowStore.ts
• src/frontend/src/utils/reactflowUtils.ts
• src/backend/base/langflow/schema/graph.py
• src/backend/base/langflow/processing/load.py
• src/backend/base/langflow/processing/process.py

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[Preet1234-ux]

@dosu

Please provide me with the full code like below :

from langflow import load_flow_from_json
TWEAKS = {
  "LLMCheckerChain-0xJIZ": {},
  "LlamaComponent-14J2Z": {}
}
flow = load_flow_from_json("Untitled document.json", tweaks=TWEAKS)
# Now you can use it like any chain
inputs = {"query":""}
flow(inputs)

[dosubot[bot]]

To optimize your langflow application for repeated queries without rebuilding the flow each time, you can implement a caching mechanism. This approach involves checking if the flow is already cached before loading or building it, thus significantly reducing the time taken for subsequent queries. Below is a Python code example that demonstrates how to incorporate caching into your application:

import json
from pathlib import Path
from langflow.graph import Graph
from langflow.services.cache.service import CacheService, ThreadingInMemoryCache

# Initialize cache service
_cache_service = CacheService(cache=ThreadingInMemoryCache())

def load_flow_from_json_with_cache(flow_path, tweaks=None):
    flow_key = str(flow_path)
    
    # Check if the flow is already in the cache
    cached_graph = _cache_service.get(flow_key)
    if cached_graph:
        return cached_graph

    # If not in cache, load the flow from JSON and cache it
    with open(flow_path, "r", encoding="utf-8") as f:
        flow_graph = json.load(f)
    # Assume tweaks processing and graph building here
    graph = Graph.from_payload(flow_graph["data"])
    _cache_service.set(flow_key, graph)
    return graph

# Example usage
flow_path = "Untitled document.json"
TWEAKS = {
  "LLMCheckerChain-0xJIZ": {},
  "LlamaComponent-14J2Z": {}
}
flow = load_flow_from_json_with_cache(flow_path, tweaks=TWEAKS)
inputs = {"query": ""}
# Use the flow with your inputs

This code snippet demonstrates how to use a caching mechanism with the CacheService and ThreadingInMemoryCache classes to avoid rebuilding the flow for each new query. The load_flow_from_json_with_cache function checks if the flow is already cached; if not, it loads the flow from the specified JSON file, processes any tweaks, builds the flow, and caches it for future use. This ensures that the flow is built only once and reused for subsequent queries, improving the efficiency of your application.

Details

Sources

• src/backend/base/langflow/processing/load.py
• src/backend/base/langflow/services/cache/init.py
• src/backend/langflow/services/cache/service.py
• src/backend/base/langflow/services/cache/service.py
• src/backend/base/langflow/services/cache/base.py
• src/backend/langflow/services/cache/factory.py
• src/backend/langflow/graph/graph/base.py
• src/backend/base/langflow/graph/graph/base.py
• src/backend/base/langflow/graph/graph/base.py
• src/backend/base/langflow/processing/process.py
• src/backend/langflow/settings.py
• src/backend/langflow/services/settings/base.py
• src/backend/base/langflow/services/settings/base.py

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[Arron-Clague]

Hi there : I would also like to know how to optimise this : I don't think the @dosu code looks correct/up to date for 1.0.14 ?

[codenprogressive]

did you consider freezing the components that you don't want to re-execute? https://docs.langflow.org/whats-new-a-new-chapter-langflow#4912d08da5464ff2aff595d6b26fd809

[Arron-Clague]

@codenprogressive I want all the components to execute : the issue is that I think the flow is rebuilt from the JSON definition each time the flow is called. However : Im on version 1.0.14 now, and I'm seeing on the docs pages, load_flow_from_json/run_flow_from_json may not be the correct way to address the endpoint for interactive messages :

def run_flow(message: str,
flow_id: str,
output_type: str = "chat",
input_type: str = "chat",
tweaks: Optional[dict] = None,
api_key: Optional[str] = "fgdgfdgdfgfdgfdgfdggdfg") -> dict:
"""Run a flow with a given message and optional tweaks.

:param message: The message to send to the flow
:param flow_id: The ID of the flow to run
:param tweaks: Optional tweaks to customize the flow
:return: The JSON response from the flow
"""
api_url = f"{BASE_API_URL}/{flow_id}"
payload = {
"input_value": message,
"output_type": output_type,
"input_type": input_type,
}
headers = None
if tweaks:
    payload["tweaks"] = tweaks
if api_key:
    headers = {"x-api-key": api_key}
response = requests.post(api_url, json=payload, headers=headers)
return response.json()