Merge branch 'main' into cz/mvp-components-flags

src/backend/base/langflow/load/load.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 from dotenv import load_dotenv
+from loguru import logger
 
 from langflow.graph import Graph
 from langflow.graph.schema import RunOutputs
@@ -43,7 +44,7 @@ def load_flow_from_json(
     """
     # If input is a file path, load JSON from the file
     log_file_path = Path(log_file) if log_file else None
-    configure(log_level=log_level, log_file=log_file_path, disable=disable_logs)
+    configure(log_level=log_level, log_file=log_file_path, disable=disable_logs, async_file=True)
 
     # override env variables with .env file
     if env_file:
@@ -119,7 +120,7 @@ async def arun_flow_from_json(
         cache=cache,
         disable_logs=disable_logs,
     )
-    return await run_graph(
+    result = await run_graph(
         graph=graph,
         session_id=session_id,
         input_value=input_value,
@@ -128,6 +129,8 @@ async def arun_flow_from_json(
         output_component=output_component,
         fallback_to_env_vars=fallback_to_env_vars,
     )
+    await logger.complete()
+    return result
 
 
 def run_flow_from_json(

src/backend/base/langflow/logging/logger.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import logging
 import os
@@ -8,7 +9,10 @@
 from typing import TypedDict
 
 import orjson
-from loguru import logger
+from loguru import _defaults, logger
+from loguru._error_interceptor import ErrorInterceptor
+from loguru._file_sink import FileSink
+from loguru._simple_sinks import AsyncSink
 from platformdirs import user_cache_dir
 from rich.logging import RichHandler
 from typing_extensions import NotRequired
@@ -136,12 +140,30 @@ class LogConfig(TypedDict):
     log_env: NotRequired[str]
 
 
+class AsyncFileSink(AsyncSink):
+    def __init__(self, file):
+        self._sink = FileSink(
+            path=file,
+            rotation="10 MB",  # Log rotation based on file size
+        )
+        super().__init__(self.write_async, None, ErrorInterceptor(_defaults.LOGURU_CATCH, -1))
+
+    async def complete(self):
+        await asyncio.to_thread(self._sink.stop)
+        for task in self._tasks:
+            await self._complete_task(task)
+
+    async def write_async(self, message):
+        await asyncio.to_thread(self._sink.write, message)
+
+
 def configure(
     *,
     log_level: str | None = None,
     log_file: Path | None = None,
     disable: bool | None = False,
     log_env: str | None = None,
+    async_file: bool = False,
 ) -> None:
     if disable and log_level is None and log_file is None:
         logger.disable("langflow")
@@ -187,14 +209,12 @@ def configure(
             log_file = cache_dir / "langflow.log"
             logger.debug(f"Log file: {log_file}")
         try:
-            log_file = Path(log_file)
             log_file.parent.mkdir(parents=True, exist_ok=True)
 
             logger.add(
-                sink=str(log_file),
+                sink=AsyncFileSink(log_file) if async_file else log_file,
                 level=log_level.upper(),
                 format=log_format,
-                rotation="10 MB",  # Log rotation based on file size
                 serialize=True,
             )
         except Exception:  # noqa: BLE001

src/backend/base/langflow/main.py

@@ -94,6 +94,7 @@ def _initialize():
 
     @asynccontextmanager
     async def lifespan(_app: FastAPI):
+        configure(async_file=True)
         # Startup message
         if version:
             rprint(f"[bold green]Starting Langflow v{version}...[/bold green]")
@@ -113,6 +114,7 @@ async def lifespan(_app: FastAPI):
         # Shutdown message
         rprint("[bold red]Shutting down Langflow...[/bold red]")
         await teardown_services()
+        await logger.complete()
 
     return lifespan
 

src/frontend/src/modals/apiModal/utils/get-curl-code.tsx

@@ -12,6 +12,7 @@ export function getCurlRunCode({
   isAuth,
   tweaksBuildedObject,
   endpointName,
+  activeTweaks,
 }: GetCodeType): string {
   let tweaksString = "{}";
   const inputs = useFlowStore.getState().inputs;
@@ -28,7 +29,7 @@ export function getCurlRunCode({
     -H 'Content-Type: application/json'\\${
       !isAuth ? `\n  -H 'x-api-key: <your api key>'\\` : ""
     }
-    -d '{"input_value": "message",
+    -d '{${!activeTweaks ? `"input_value": "message",` : ""}
     "output_type": ${hasChatOutput ? '"chat"' : '"text"'},
     "input_type": ${hasChatInput ? '"chat"' : '"text"'},
     "tweaks": ${tweaksString}}'

src/frontend/src/modals/apiModal/utils/get-js-api-code.tsx

@@ -13,6 +13,7 @@ export default function getJsApiCode({
   isAuth,
   tweaksBuildedObject,
   endpointName,
+  activeTweaks,
 }: GetCodeType): string {
   let tweaksString = "{}";
   if (tweaksBuildedObject)
@@ -23,7 +24,7 @@ export default function getJsApiCode({
         this.baseURL = baseURL;
         this.apiKey = apiKey;
     }
-  
+
     async post(endpoint, body, headers = {"Content-Type": "application/json"}) {
       if (this.apiKey) {
             headers["Authorization"] = \`Bearer \${this.apiKey}\`;
@@ -35,7 +36,7 @@ export default function getJsApiCode({
                 headers: headers,
                 body: JSON.stringify(body)
             });
-  
+
             const responseMessage = await response.json();
             if (!response.ok) {
                 throw new Error(\`\${response.status} \${response.statusText} - \${JSON.stringify(responseMessage)}\`);
@@ -46,12 +47,12 @@ export default function getJsApiCode({
             throw error;
         }
     }
-  
+
     async initiateSession(flowId, inputValue, inputType = 'chat', outputType = 'chat', stream = false, tweaks = {}) {
         const endpoint = \`/api/v1/run/\${flowId}?stream=\${stream}\`;
-        return this.post(endpoint, { input_value: inputValue, input_type: inputType, output_type: outputType, tweaks: tweaks });
+        return this.post(endpoint, { ${activeTweaks ? "" : "input_value: inputValue, "}input_type: inputType, output_type: outputType, tweaks: tweaks });
     }
-  
+
     async handleStream(streamUrl, onUpdate, onClose, onError) {
       try {
         const response = await fetch(streamUrl);
@@ -66,7 +67,7 @@ export default function getJsApiCode({
           }
           const chunk = decoder.decode(value);
           const lines = chunk.split(\'\\n\').filter(line => line.trim() !== '');
-          
+
           for (const line of lines) {
             if (line.startsWith('data: ')) {
               try {
@@ -83,7 +84,7 @@ export default function getJsApiCode({
         onError(error);
       }
     }
-  
+
     async runFlow(flowIdOrName, inputValue, inputType = 'chat', outputType = 'chat', tweaks, stream = false, onUpdate, onClose, onError) {
         try {
             const initResponse = await this.initiateSession(flowIdOrName, inputValue, inputType, outputType, stream, tweaks);
@@ -98,13 +99,13 @@ export default function getJsApiCode({
         }
     }
   }
-  
+
   async function main(inputValue, inputType = 'chat', outputType = 'chat', stream = false) {
     const flowIdOrName = '${endpointName || flowId}';
     const langflowClient = new LangflowClient('${window.location.protocol}//${window.location.host}',
           ${isAuth ? "'your-api-key'" : "null"});
     const tweaks = ${tweaksString};
-  
+
     try {
         const response = await langflowClient.runFlow(
             flowIdOrName,
@@ -117,19 +118,19 @@ export default function getJsApiCode({
             (message) => console.log("Stream Closed:", message), // onClose
             (error) => console.error("Stream Error:", error) // onError
         );
-  
+
         if (!stream && response) {
             const flowOutputs = response.outputs[0];
             const firstComponentOutputs = flowOutputs.outputs[0];
             const output = firstComponentOutputs.outputs.message;
-  
+
             console.log("Final Output:", output.message.text);
         }
     } catch (error) {
         console.error('Main Error:', error.message);
     }
   }
-  
+
   const args = process.argv.slice(2);
   main(
     args[0], // inputValue

src/frontend/src/modals/apiModal/utils/get-python-api-code.tsx

@@ -12,6 +12,7 @@ export default function getPythonApiCode({
   flowId,
   tweaksBuildedObject,
   endpointName,
+  activeTweaks,
 }: GetCodeType): string {
   let tweaksString = "{}";
   if (tweaksBuildedObject)
@@ -61,7 +62,7 @@ def run_flow(message: str,
     api_url = f"{BASE_API_URL}/api/v1/run/{endpoint}"
 
     payload = {
-        "input_value": message,
+        ${!activeTweaks ? `"input_value": message,` : ""}
         "output_type": output_type,
         "input_type": input_type,
     }

src/frontend/src/modals/apiModal/utils/get-python-code.tsx

@@ -9,6 +9,7 @@ import { GetCodeType } from "@/types/tweaks";
 export default function getPythonCode({
   flowName,
   tweaksBuildedObject,
+  activeTweaks,
 }: GetCodeType): string {
   let tweaksString = "{}";
   if (tweaksBuildedObject)
@@ -21,8 +22,7 @@ export default function getPythonCode({
 TWEAKS = ${tweaksString}
 
 result = run_flow_from_json(flow="${flowName}.json",
-                            input_value="message",
-                            session_id="", # provide a session id if you want to use session state
+                            ${!activeTweaks ? `input_value="message",\n                            ` : ""}session_id="", # provide a session id if you want to use session state
                             fallback_to_env_vars=True, # False by default
                             tweaks=TWEAKS)`;
 }

src/frontend/src/stores/tweaksStore.ts

@@ -98,6 +98,7 @@ export const useTweaksStore = create<TweaksStoreType>((set, get) => ({
       isAuth: autoLogin,
       tweaksBuildedObject: tweak,
       endpointName: flow?.endpoint_name,
+      activeTweaks: get().activeTweaks,
     };
 
     if (getCodes) {

src/frontend/src/types/tweaks/index.ts

@@ -13,4 +13,5 @@ export type GetCodeType = {
   isAuth: boolean;
   tweaksBuildedObject: {};
   endpointName?: string | null;
+  activeTweaks: boolean;
 };