|
|
from langflow.custom import Component |
|
|
from langflow.io import BoolInput, DropdownInput, IntInput, MessageInput, MessageTextInput, Output |
|
|
from langflow.schema.message import Message |
|
|
|
|
|
|
|
|
class ConditionalRouterComponent(Component): |
|
|
display_name = "If-Else" |
|
|
description = "Routes an input message to a corresponding output based on text comparison." |
|
|
icon = "split" |
|
|
name = "ConditionalRouter" |
|
|
|
|
|
def __init__(self, *args, **kwargs): |
|
|
super().__init__(*args, **kwargs) |
|
|
self.__iteration_updated = False |
|
|
|
|
|
inputs = [ |
|
|
MessageTextInput( |
|
|
name="input_text", |
|
|
display_name="Text Input", |
|
|
info="The primary text input for the operation.", |
|
|
), |
|
|
MessageTextInput( |
|
|
name="match_text", |
|
|
display_name="Match Text", |
|
|
info="The text input to compare against.", |
|
|
), |
|
|
DropdownInput( |
|
|
name="operator", |
|
|
display_name="Operator", |
|
|
options=["equals", "not equals", "contains", "starts with", "ends with"], |
|
|
info="The operator to apply for comparing the texts.", |
|
|
value="equals", |
|
|
), |
|
|
BoolInput( |
|
|
name="case_sensitive", |
|
|
display_name="Case Sensitive", |
|
|
info="If true, the comparison will be case sensitive.", |
|
|
value=False, |
|
|
advanced=True, |
|
|
), |
|
|
MessageInput( |
|
|
name="message", |
|
|
display_name="Message", |
|
|
info="The message to pass through either route.", |
|
|
advanced=True, |
|
|
), |
|
|
IntInput( |
|
|
name="max_iterations", |
|
|
display_name="Max Iterations", |
|
|
info="The maximum number of iterations for the conditional router.", |
|
|
value=10, |
|
|
), |
|
|
DropdownInput( |
|
|
name="default_route", |
|
|
display_name="Default Route", |
|
|
options=["true_result", "false_result"], |
|
|
info="The default route to take when max iterations are reached.", |
|
|
value="false_result", |
|
|
advanced=True, |
|
|
), |
|
|
] |
|
|
|
|
|
outputs = [ |
|
|
Output(display_name="True", name="true_result", method="true_response"), |
|
|
Output(display_name="False", name="false_result", method="false_response"), |
|
|
] |
|
|
|
|
|
def _pre_run_setup(self): |
|
|
self.__iteration_updated = False |
|
|
|
|
|
def evaluate_condition(self, input_text: str, match_text: str, operator: str, *, case_sensitive: bool) -> bool: |
|
|
if not case_sensitive: |
|
|
input_text = input_text.lower() |
|
|
match_text = match_text.lower() |
|
|
|
|
|
if operator == "equals": |
|
|
return input_text == match_text |
|
|
if operator == "not equals": |
|
|
return input_text != match_text |
|
|
if operator == "contains": |
|
|
return match_text in input_text |
|
|
if operator == "starts with": |
|
|
return input_text.startswith(match_text) |
|
|
if operator == "ends with": |
|
|
return input_text.endswith(match_text) |
|
|
return False |
|
|
|
|
|
def iterate_and_stop_once(self, route_to_stop: str): |
|
|
if not self.__iteration_updated: |
|
|
self.update_ctx({f"{self._id}_iteration": self.ctx.get(f"{self._id}_iteration", 0) + 1}) |
|
|
self.__iteration_updated = True |
|
|
if self.ctx.get(f"{self._id}_iteration", 0) >= self.max_iterations and route_to_stop == self.default_route: |
|
|
|
|
|
route_to_stop = "true_result" if route_to_stop == "false_result" else "false_result" |
|
|
self.stop(route_to_stop) |
|
|
|
|
|
def true_response(self) -> Message | str: |
|
|
result = self.evaluate_condition( |
|
|
self.input_text, self.match_text, self.operator, case_sensitive=self.case_sensitive |
|
|
) |
|
|
if result: |
|
|
self.status = self.message |
|
|
self.iterate_and_stop_once("false_result") |
|
|
return self.message |
|
|
self.iterate_and_stop_once("true_result") |
|
|
return "" |
|
|
|
|
|
def false_response(self) -> Message | str: |
|
|
result = self.evaluate_condition( |
|
|
self.input_text, self.match_text, self.operator, case_sensitive=self.case_sensitive |
|
|
) |
|
|
if not result: |
|
|
self.status = self.message |
|
|
self.iterate_and_stop_once("true_result") |
|
|
return self.message |
|
|
self.iterate_and_stop_once("false_result") |
|
|
return "" |
|
|
|
