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

Refactor invoke_lc_runnable function parameters

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-02-07 14:07:02 -03:00
commit ee43952e42
1 changed files with 7 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/backend/langflow/graph/vertex/utils.py
View file

@ -11,7 +11,7 @@ def is_basic_type(obj):
async def invoke_lc_runnable( async def invoke_lc_runnable(
langchain_object: Runnable, inputs: dict, has_external_output: bool, session_id: Optional[str] = None, **kwargs built_object: Runnable, inputs: dict, has_external_output: bool, session_id: Optional[str] = None, **kwargs
) -> Union[str, BaseMessage]: ) -> Union[str, BaseMessage]:
# Setup callbacks for asynchronous execution # Setup callbacks for asynchronous execution
from langflow.processing.base import setup_callbacks from langflow.processing.base import setup_callbacks
@ -19,15 +19,15 @@ async def invoke_lc_runnable(
callbacks = setup_callbacks(sync=False, trace_id=session_id, **kwargs) callbacks = setup_callbacks(sync=False, trace_id=session_id, **kwargs)
try: try:
if has_external_output and hasattr(langchain_object, "astream"): if has_external_output and hasattr(built_object, "astream"):
# Asynchronous stream handling if supported and required # Asynchronous stream handling if supported and required
output = "" output = ""
async for chunk in langchain_object.astream(inputs, {"callbacks": callbacks}): async for chunk in built_object.astream(inputs, {"callbacks": callbacks}):
output += chunk output += chunk
return output return output
else: else:
# Direct asynchronous invocation # Direct asynchronous invocation
return await langchain_object.ainvoke(inputs, {"callbacks": callbacks}) return await built_object.ainvoke(inputs, {"callbacks": callbacks})
except Exception as async_exc: except Exception as async_exc:
logger.debug(f"Async error, falling back to sync: {str(async_exc)}") logger.debug(f"Async error, falling back to sync: {str(async_exc)}")
@ -35,15 +35,15 @@ async def invoke_lc_runnable(
sync_callbacks = setup_callbacks(sync=True, trace_id=session_id, **kwargs) sync_callbacks = setup_callbacks(sync=True, trace_id=session_id, **kwargs)
try: try:
# Synchronous fallback if asynchronous execution fails # Synchronous fallback if asynchronous execution fails
if has_external_output and hasattr(langchain_object, "stream"): if has_external_output and hasattr(built_object, "stream"):
# Synchronous stream handling if supported and required # Synchronous stream handling if supported and required
output = "" output = ""
for chunk in langchain_object.stream(inputs, {"callbacks": sync_callbacks}): for chunk in built_object.stream(inputs, {"callbacks": sync_callbacks}):
output += chunk output += chunk
return output return output
else: else:
# Direct synchronous invocation # Direct synchronous invocation
return langchain_object.invoke(inputs, {"callbacks": sync_callbacks}) return built_object.invoke(inputs, {"callbacks": sync_callbacks})
except Exception as sync_exc: except Exception as sync_exc:
logger.error(f"Sync error after async failure: {str(sync_exc)}") logger.error(f"Sync error after async failure: {str(sync_exc)}")
# Handle or re-raise exception as appropriate for your application # Handle or re-raise exception as appropriate for your application