refactor: update dependency versions and streamline backend installation commands (#3939)
* Update Makefile to streamline backend dependency installation commands * Update dependency versions in pyproject.toml for weaviate-client, httpx, and others * Update dependency versions in pyproject.toml for better compatibility and stability * new lock * refactor: streamline backend dependency installation commands * update examples formatting
This commit is contained in:
Gabriel Luiz Freitas Almeida
GitHub
parent
b877bc9e08
commit
bf2aadf6a4
9 changed files with 73 additions and 60 deletions
|
|
@ -1211,7 +1211,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "import ast\nimport operator\nfrom typing import List\n\nfrom langchain.tools import StructuredTool\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import MessageTextInput\nfrom langflow.schema import Data\n\n\nclass CalculatorToolComponent(LCToolComponent):\n display_name = \"Calculator\"\n description = \"Perform basic arithmetic operations on a given expression.\"\n icon = \"calculator\"\n name = \"CalculatorTool\"\n\n inputs = [\n MessageTextInput(\n name=\"expression\",\n display_name=\"Expression\",\n info=\"The arithmetic expression to evaluate (e.g., '4*4*(33/22)+12-20').\",\n ),\n ]\n\n class CalculatorToolSchema(BaseModel):\n expression: str = Field(..., description=\"The arithmetic expression to evaluate.\")\n\n def run_model(self) -> List[Data]:\n return self._evaluate_expression(self.expression)\n\n def build_tool(self) -> Tool:\n return StructuredTool.from_function(\n name=\"calculator\",\n description=\"Evaluate basic arithmetic expressions. Input should be a string containing the expression.\",\n func=self._evaluate_expression,\n args_schema=self.CalculatorToolSchema,\n )\n\n def _evaluate_expression(self, expression: str) -> List[Data]:\n try:\n # Define the allowed operators\n operators = {\n ast.Add: operator.add,\n ast.Sub: operator.sub,\n ast.Mult: operator.mul,\n ast.Div: operator.truediv,\n ast.Pow: operator.pow,\n }\n\n def eval_expr(node):\n if isinstance(node, ast.Num):\n return node.n\n elif isinstance(node, ast.BinOp):\n return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))\n elif isinstance(node, ast.UnaryOp):\n return operators[type(node.op)](eval_expr(node.operand))\n else:\n raise TypeError(node)\n\n # Parse the expression and evaluate it\n tree = ast.parse(expression, mode=\"eval\")\n result = eval_expr(tree.body)\n\n # Format the result to a reasonable number of decimal places\n formatted_result = f\"{result:.6f}\".rstrip(\"0\").rstrip(\".\")\n\n self.status = formatted_result\n return [Data(data={\"result\": formatted_result})]\n\n except (SyntaxError, TypeError, KeyError) as e:\n error_message = f\"Invalid expression: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except ZeroDivisionError:\n error_message = \"Error: Division by zero\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except Exception as e:\n error_message = f\"Error: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n"
|
||||
"value": "import ast\nimport operator\n\nfrom langchain.tools import StructuredTool\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import MessageTextInput\nfrom langflow.schema import Data\n\n\nclass CalculatorToolComponent(LCToolComponent):\n display_name = \"Calculator\"\n description = \"Perform basic arithmetic operations on a given expression.\"\n icon = \"calculator\"\n name = \"CalculatorTool\"\n\n inputs = [\n MessageTextInput(\n name=\"expression\",\n display_name=\"Expression\",\n info=\"The arithmetic expression to evaluate (e.g., '4*4*(33/22)+12-20').\",\n ),\n ]\n\n class CalculatorToolSchema(BaseModel):\n expression: str = Field(..., description=\"The arithmetic expression to evaluate.\")\n\n def run_model(self) -> list[Data]:\n return self._evaluate_expression(self.expression)\n\n def build_tool(self) -> Tool:\n return StructuredTool.from_function(\n name=\"calculator\",\n description=\"Evaluate basic arithmetic expressions. Input should be a string containing the expression.\",\n func=self._evaluate_expression,\n args_schema=self.CalculatorToolSchema,\n )\n\n def _evaluate_expression(self, expression: str) -> list[Data]:\n try:\n # Define the allowed operators\n operators = {\n ast.Add: operator.add,\n ast.Sub: operator.sub,\n ast.Mult: operator.mul,\n ast.Div: operator.truediv,\n ast.Pow: operator.pow,\n }\n\n def eval_expr(node):\n if isinstance(node, ast.Num):\n return node.n\n elif isinstance(node, ast.BinOp):\n return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))\n elif isinstance(node, ast.UnaryOp):\n return operators[type(node.op)](eval_expr(node.operand))\n else:\n raise TypeError(node)\n\n # Parse the expression and evaluate it\n tree = ast.parse(expression, mode=\"eval\")\n result = eval_expr(tree.body)\n\n # Format the result to a reasonable number of decimal places\n formatted_result = f\"{result:.6f}\".rstrip(\"0\").rstrip(\".\")\n\n self.status = formatted_result\n return [Data(data={\"result\": formatted_result})]\n\n except (SyntaxError, TypeError, KeyError) as e:\n error_message = f\"Invalid expression: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except ZeroDivisionError:\n error_message = \"Error: Division by zero\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except Exception as e:\n error_message = f\"Error: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n"
|
||||
},
|
||||
"expression": {
|
||||
"_input_type": "MessageTextInput",
|
||||
|
|
@ -1321,7 +1321,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "import importlib\nfrom typing import List, Union\n\nfrom langchain.tools import StructuredTool\nfrom langchain_experimental.utilities import PythonREPL\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import StrInput\nfrom langflow.schema import Data\n\n\nclass PythonREPLToolComponent(LCToolComponent):\n display_name = \"Python REPL Tool\"\n description = \"A tool for running Python code in a REPL environment.\"\n name = \"PythonREPLTool\"\n\n inputs = [\n StrInput(\n name=\"name\",\n display_name=\"Tool Name\",\n info=\"The name of the tool.\",\n value=\"python_repl\",\n ),\n StrInput(\n name=\"description\",\n display_name=\"Tool Description\",\n info=\"A description of the tool.\",\n value=\"A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.\",\n ),\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy'.\",\n value=\"math\",\n ),\n StrInput(\n name=\"code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute.\",\n value=\"print('Hello, World!')\",\n ),\n ]\n\n class PythonREPLSchema(BaseModel):\n code: str = Field(..., description=\"The Python code to execute.\")\n\n def get_globals(self, global_imports: Union[str, List[str]]) -> dict:\n global_dict = {}\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n raise ValueError(\"global_imports must be either a string or a list\")\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError:\n raise ImportError(f\"Could not import module {module}\")\n return global_dict\n\n def build_tool(self) -> Tool:\n _globals = self.get_globals(self.global_imports)\n python_repl = PythonREPL(_globals=_globals)\n\n def run_python_code(code: str) -> str:\n try:\n return python_repl.run(code)\n except Exception as e:\n return f\"Error: {str(e)}\"\n\n tool = StructuredTool.from_function(\n name=self.name,\n description=self.description,\n func=run_python_code,\n args_schema=self.PythonREPLSchema,\n )\n\n self.status = f\"Python REPL Tool created with global imports: {self.global_imports}\"\n return tool\n\n def run_model(self) -> List[Data]:\n tool = self.build_tool()\n result = tool.run(self.code)\n return [Data(data={\"result\": result})]\n"
|
||||
"value": "import importlib\n\nfrom langchain.tools import StructuredTool\nfrom langchain_experimental.utilities import PythonREPL\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import StrInput\nfrom langflow.schema import Data\n\n\nclass PythonREPLToolComponent(LCToolComponent):\n display_name = \"Python REPL Tool\"\n description = \"A tool for running Python code in a REPL environment.\"\n name = \"PythonREPLTool\"\n\n inputs = [\n StrInput(\n name=\"name\",\n display_name=\"Tool Name\",\n info=\"The name of the tool.\",\n value=\"python_repl\",\n ),\n StrInput(\n name=\"description\",\n display_name=\"Tool Description\",\n info=\"A description of the tool.\",\n value=\"A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.\",\n ),\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy'.\",\n value=\"math\",\n ),\n StrInput(\n name=\"code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute.\",\n value=\"print('Hello, World!')\",\n ),\n ]\n\n class PythonREPLSchema(BaseModel):\n code: str = Field(..., description=\"The Python code to execute.\")\n\n def get_globals(self, global_imports: str | list[str]) -> dict:\n global_dict = {}\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n raise ValueError(\"global_imports must be either a string or a list\")\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError:\n raise ImportError(f\"Could not import module {module}\")\n return global_dict\n\n def build_tool(self) -> Tool:\n _globals = self.get_globals(self.global_imports)\n python_repl = PythonREPL(_globals=_globals)\n\n def run_python_code(code: str) -> str:\n try:\n return python_repl.run(code)\n except Exception as e:\n return f\"Error: {str(e)}\"\n\n tool = StructuredTool.from_function(\n name=self.name,\n description=self.description,\n func=run_python_code,\n args_schema=self.PythonREPLSchema,\n )\n\n self.status = f\"Python REPL Tool created with global imports: {self.global_imports}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n result = tool.run(self.code)\n return [Data(data={\"result\": result})]\n"
|
||||
},
|
||||
"description": {
|
||||
"_input_type": "StrInput",
|
||||
|
|
|
|||
|
|
@ -4109,7 +4109,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "from typing import Any, Dict, List, Optional\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: Optional[Dict[str, Any]] = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: Optional[Dict[str, Any]] = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> List[Dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> List[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
"value": "from typing import Any\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: dict[str, Any] | None = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: dict[str, Any] | None = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> list[dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
},
|
||||
"engine": {
|
||||
"advanced": false,
|
||||
|
|
|
|||
|
|
@ -2668,7 +2668,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "from typing import Any, Dict, List, Optional\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: Optional[Dict[str, Any]] = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: Optional[Dict[str, Any]] = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> List[Dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> List[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
"value": "from typing import Any\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: dict[str, Any] | None = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: dict[str, Any] | None = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> list[dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
},
|
||||
"engine": {
|
||||
"advanced": false,
|
||||
|
|
|
|||
|
|
@ -1469,7 +1469,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "from typing import Any, Dict, List, Optional\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: Optional[Dict[str, Any]] = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: Optional[Dict[str, Any]] = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> List[Dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> List[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
"value": "from typing import Any\n\nfrom langchain.tools import StructuredTool\nfrom langchain_community.utilities.searchapi import SearchApiAPIWrapper\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import DictInput, IntInput, MessageTextInput, MultilineInput, SecretStrInput\nfrom langflow.schema import Data\n\n\nclass SearchAPIComponent(LCToolComponent):\n display_name: str = \"Search API\"\n description: str = \"Call the searchapi.io API with result limiting\"\n name = \"SearchAPI\"\n documentation: str = \"https://www.searchapi.io/docs/google\"\n\n inputs = [\n MessageTextInput(name=\"engine\", display_name=\"Engine\", value=\"google\"),\n SecretStrInput(name=\"api_key\", display_name=\"SearchAPI API Key\", required=True),\n MultilineInput(\n name=\"input_value\",\n display_name=\"Input\",\n ),\n DictInput(name=\"search_params\", display_name=\"Search parameters\", advanced=True, is_list=True),\n IntInput(name=\"max_results\", display_name=\"Max Results\", value=5, advanced=True),\n IntInput(name=\"max_snippet_length\", display_name=\"Max Snippet Length\", value=100, advanced=True),\n ]\n\n class SearchAPISchema(BaseModel):\n query: str = Field(..., description=\"The search query\")\n params: dict[str, Any] | None = Field(default_factory=dict, description=\"Additional search parameters\")\n max_results: int = Field(5, description=\"Maximum number of results to return\")\n max_snippet_length: int = Field(100, description=\"Maximum length of each result snippet\")\n\n def _build_wrapper(self):\n return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)\n\n def build_tool(self) -> Tool:\n wrapper = self._build_wrapper()\n\n def search_func(\n query: str, params: dict[str, Any] | None = None, max_results: int = 5, max_snippet_length: int = 100\n ) -> list[dict[str, Any]]:\n params = params or {}\n full_results = wrapper.results(query=query, **params)\n organic_results = full_results.get(\"organic_results\", [])[:max_results]\n\n limited_results = []\n for result in organic_results:\n limited_result = {\n \"title\": result.get(\"title\", \"\")[:max_snippet_length],\n \"link\": result.get(\"link\", \"\"),\n \"snippet\": result.get(\"snippet\", \"\")[:max_snippet_length],\n }\n limited_results.append(limited_result)\n\n return limited_results\n\n tool = StructuredTool.from_function(\n name=\"search_api\",\n description=\"Search for recent results using searchapi.io with result limiting\",\n func=search_func,\n args_schema=self.SearchAPISchema,\n )\n\n self.status = f\"Search API Tool created with engine: {self.engine}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n results = tool.run(\n {\n \"query\": self.input_value,\n \"params\": self.search_params or {},\n \"max_results\": self.max_results,\n \"max_snippet_length\": self.max_snippet_length,\n }\n )\n\n data_list = [Data(data=result, text=result.get(\"snippet\", \"\")) for result in results]\n\n self.status = data_list\n return data_list\n"
|
||||
},
|
||||
"engine": {
|
||||
"_input_type": "MessageTextInput",
|
||||
|
|
@ -2310,7 +2310,7 @@
|
|||
"show": true,
|
||||
"title_case": false,
|
||||
"type": "code",
|
||||
"value": "import ast\nimport operator\nfrom typing import List\n\nfrom langchain.tools import StructuredTool\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import MessageTextInput\nfrom langflow.schema import Data\n\n\nclass CalculatorToolComponent(LCToolComponent):\n display_name = \"Calculator\"\n description = \"Perform basic arithmetic operations on a given expression.\"\n icon = \"calculator\"\n name = \"CalculatorTool\"\n\n inputs = [\n MessageTextInput(\n name=\"expression\",\n display_name=\"Expression\",\n info=\"The arithmetic expression to evaluate (e.g., '4*4*(33/22)+12-20').\",\n ),\n ]\n\n class CalculatorToolSchema(BaseModel):\n expression: str = Field(..., description=\"The arithmetic expression to evaluate.\")\n\n def run_model(self) -> List[Data]:\n return self._evaluate_expression(self.expression)\n\n def build_tool(self) -> Tool:\n return StructuredTool.from_function(\n name=\"calculator\",\n description=\"Evaluate basic arithmetic expressions. Input should be a string containing the expression.\",\n func=self._evaluate_expression,\n args_schema=self.CalculatorToolSchema,\n )\n\n def _evaluate_expression(self, expression: str) -> List[Data]:\n try:\n # Define the allowed operators\n operators = {\n ast.Add: operator.add,\n ast.Sub: operator.sub,\n ast.Mult: operator.mul,\n ast.Div: operator.truediv,\n ast.Pow: operator.pow,\n }\n\n def eval_expr(node):\n if isinstance(node, ast.Num):\n return node.n\n elif isinstance(node, ast.BinOp):\n return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))\n elif isinstance(node, ast.UnaryOp):\n return operators[type(node.op)](eval_expr(node.operand))\n else:\n raise TypeError(node)\n\n # Parse the expression and evaluate it\n tree = ast.parse(expression, mode=\"eval\")\n result = eval_expr(tree.body)\n\n # Format the result to a reasonable number of decimal places\n formatted_result = f\"{result:.6f}\".rstrip(\"0\").rstrip(\".\")\n\n self.status = formatted_result\n return [Data(data={\"result\": formatted_result})]\n\n except (SyntaxError, TypeError, KeyError) as e:\n error_message = f\"Invalid expression: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except ZeroDivisionError:\n error_message = \"Error: Division by zero\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except Exception as e:\n error_message = f\"Error: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n"
|
||||
"value": "import ast\nimport operator\n\nfrom langchain.tools import StructuredTool\nfrom pydantic import BaseModel, Field\n\nfrom langflow.base.langchain_utilities.model import LCToolComponent\nfrom langflow.field_typing import Tool\nfrom langflow.inputs import MessageTextInput\nfrom langflow.schema import Data\n\n\nclass CalculatorToolComponent(LCToolComponent):\n display_name = \"Calculator\"\n description = \"Perform basic arithmetic operations on a given expression.\"\n icon = \"calculator\"\n name = \"CalculatorTool\"\n\n inputs = [\n MessageTextInput(\n name=\"expression\",\n display_name=\"Expression\",\n info=\"The arithmetic expression to evaluate (e.g., '4*4*(33/22)+12-20').\",\n ),\n ]\n\n class CalculatorToolSchema(BaseModel):\n expression: str = Field(..., description=\"The arithmetic expression to evaluate.\")\n\n def run_model(self) -> list[Data]:\n return self._evaluate_expression(self.expression)\n\n def build_tool(self) -> Tool:\n return StructuredTool.from_function(\n name=\"calculator\",\n description=\"Evaluate basic arithmetic expressions. Input should be a string containing the expression.\",\n func=self._evaluate_expression,\n args_schema=self.CalculatorToolSchema,\n )\n\n def _evaluate_expression(self, expression: str) -> list[Data]:\n try:\n # Define the allowed operators\n operators = {\n ast.Add: operator.add,\n ast.Sub: operator.sub,\n ast.Mult: operator.mul,\n ast.Div: operator.truediv,\n ast.Pow: operator.pow,\n }\n\n def eval_expr(node):\n if isinstance(node, ast.Num):\n return node.n\n elif isinstance(node, ast.BinOp):\n return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))\n elif isinstance(node, ast.UnaryOp):\n return operators[type(node.op)](eval_expr(node.operand))\n else:\n raise TypeError(node)\n\n # Parse the expression and evaluate it\n tree = ast.parse(expression, mode=\"eval\")\n result = eval_expr(tree.body)\n\n # Format the result to a reasonable number of decimal places\n formatted_result = f\"{result:.6f}\".rstrip(\"0\").rstrip(\".\")\n\n self.status = formatted_result\n return [Data(data={\"result\": formatted_result})]\n\n except (SyntaxError, TypeError, KeyError) as e:\n error_message = f\"Invalid expression: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except ZeroDivisionError:\n error_message = \"Error: Division by zero\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n except Exception as e:\n error_message = f\"Error: {str(e)}\"\n self.status = error_message\n return [Data(data={\"error\": error_message})]\n"
|
||||
},
|
||||
"expression": {
|
||||
"_input_type": "MessageTextInput",
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
|
@ -166,7 +166,7 @@ readme = "README.md"
|
|||
|
||||
dependencies = [
|
||||
"fastapi>=0.111.0",
|
||||
"httpx",
|
||||
"httpx>=0.27",
|
||||
"uvicorn>=0.30.0",
|
||||
"gunicorn>=22.0.0",
|
||||
"langchain~=0.2.0",
|
||||
|
|
@ -178,7 +178,6 @@ dependencies = [
|
|||
"langchain-experimental>=0.0.61",
|
||||
"pydantic>=2.7.0",
|
||||
"pydantic-settings>=2.2.0",
|
||||
"websockets",
|
||||
"typer>=0.12.0",
|
||||
"cachetools>=5.3.1",
|
||||
"platformdirs>=4.2.0",
|
||||
|
|
@ -224,22 +223,22 @@ dependencies = [
|
|||
# Optional dependencies for uv
|
||||
[project.optional-dependencies]
|
||||
deploy = [
|
||||
"celery",
|
||||
"redis",
|
||||
"flower"
|
||||
"celery>=5.3.1",
|
||||
"redis>=4.6.0",
|
||||
"flower>=1.0.0"
|
||||
]
|
||||
local = [
|
||||
"llama-cpp-python",
|
||||
"sentence-transformers",
|
||||
"ctransformers"
|
||||
"llama-cpp-python>=0.2.0",
|
||||
"sentence-transformers>=2.0.0",
|
||||
"ctransformers>=0.2"
|
||||
]
|
||||
all = [
|
||||
"celery",
|
||||
"redis",
|
||||
"flower",
|
||||
"llama-cpp-python",
|
||||
"sentence-transformers",
|
||||
"ctransformers"
|
||||
"celery>=5.3.1",
|
||||
"redis>=4.6.0",
|
||||
"flower>=1.0.0",
|
||||
"llama-cpp-python>=0.2.0",
|
||||
"sentence-transformers>=2.0.0",
|
||||
"ctransformers>=0.2"
|
||||
]
|
||||
|
||||
# Development dependencies
|
||||
|
|
@ -248,7 +247,7 @@ dev = [
|
|||
"ipykernel>=6.29.0",
|
||||
"mypy>=1.11.0",
|
||||
"ruff>=0.4.5",
|
||||
"httpx",
|
||||
"httpx>=0.27",
|
||||
"pytest>=8.2.0",
|
||||
"types-requests>=2.32.0",
|
||||
"requests>=2.32.0",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue