e768098d0e
Flake8 Lint / flake8 (push) Waiting to run
Spell check CI / Spell_Check (push) Waiting to run
tools_continuous_delivery / Private PyPI non-main branch release (push) Has been skipped
tools_continuous_delivery / Private PyPI main branch release (push) Failing after 2m42s
Publish Promptflow Doc / Build (push) Has been cancelled
Publish Promptflow Doc / Deploy (push) Has been cancelled
137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""
|
|
This file can generate a meta file for the given prompt template or a python file.
|
|
"""
|
|
import inspect
|
|
import types
|
|
from dataclasses import asdict
|
|
|
|
from utils.tool_utils import function_to_interface
|
|
|
|
from promptflow.contracts.tool import Tool, ToolType
|
|
# Avoid circular dependencies: Use import 'from promptflow._internal' instead of 'from promptflow'
|
|
# since the code here is in promptflow namespace as well
|
|
from promptflow._internal import ToolProvider
|
|
from promptflow.exceptions import ErrorTarget, UserErrorException
|
|
|
|
|
|
def asdict_without_none(obj):
|
|
return asdict(obj, dict_factory=lambda x: {k: v for (k, v) in x if v})
|
|
|
|
|
|
def asdict_with_advanced_features_without_none(obj, **advanced_features):
|
|
dict_without_none = asdict_without_none(obj)
|
|
dict_without_none.update({k: v for k, v in advanced_features.items() if v})
|
|
return dict_without_none
|
|
|
|
|
|
def is_tool(f):
|
|
if not isinstance(f, types.FunctionType):
|
|
return False
|
|
if not hasattr(f, "__tool"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def collect_tool_functions_in_module(m):
|
|
tools = []
|
|
for _, obj in inspect.getmembers(m):
|
|
if is_tool(obj):
|
|
# Note that the tool should be in defined in exec but not imported in exec,
|
|
# so it should also have the same module with the current function.
|
|
if getattr(obj, "__module__", "") != m.__name__:
|
|
continue
|
|
tools.append(obj)
|
|
return tools
|
|
|
|
|
|
def collect_tool_methods_in_module(m):
|
|
tools = []
|
|
for _, obj in inspect.getmembers(m):
|
|
if isinstance(obj, type) and issubclass(obj, ToolProvider) and obj.__module__ == m.__name__:
|
|
for _, method in inspect.getmembers(obj):
|
|
if is_tool(method):
|
|
initialize_inputs = obj.get_initialize_inputs()
|
|
tools.append((method, initialize_inputs))
|
|
return tools
|
|
|
|
|
|
def _parse_tool_from_function(f, initialize_inputs=None, tool_type=ToolType.PYTHON, name=None, description=None):
|
|
if hasattr(f, "__tool") and isinstance(f.__tool, Tool):
|
|
return f.__tool
|
|
if hasattr(f, "__original_function"):
|
|
f = f.__original_function
|
|
try:
|
|
inputs, _, _ = function_to_interface(f, tool_type=tool_type, initialize_inputs=initialize_inputs)
|
|
except Exception as e:
|
|
raise BadFunctionInterface(f"Failed to parse interface for tool {f.__name__}, reason: {e}") from e
|
|
class_name = None
|
|
if "." in f.__qualname__:
|
|
class_name = f.__qualname__.replace(f".{f.__name__}", "")
|
|
# Construct the Tool structure
|
|
return Tool(
|
|
name=name or f.__qualname__,
|
|
description=description or inspect.getdoc(f),
|
|
inputs=inputs,
|
|
type=tool_type,
|
|
class_name=class_name,
|
|
function=f.__name__,
|
|
module=f.__module__,
|
|
)
|
|
|
|
|
|
def generate_python_tools_in_module(module, name, description):
|
|
tool_functions = collect_tool_functions_in_module(module)
|
|
tool_methods = collect_tool_methods_in_module(module)
|
|
return [_parse_tool_from_function(f, name=name, description=description) for f in tool_functions] + [
|
|
_parse_tool_from_function(f, initialize_inputs, name=name, description=description)
|
|
for (f, initialize_inputs) in tool_methods
|
|
]
|
|
|
|
|
|
def generate_python_tools_in_module_as_dict(module, name=None, description=None, **advanced_features):
|
|
tools = generate_python_tools_in_module(module, name, description)
|
|
return _construct_tool_dict(tools, **advanced_features)
|
|
|
|
|
|
def generate_custom_llm_tools_in_module(module, name, description):
|
|
tool_functions = collect_tool_functions_in_module(module)
|
|
tool_methods = collect_tool_methods_in_module(module)
|
|
return [
|
|
_parse_tool_from_function(f, tool_type=ToolType.CUSTOM_LLM, name=name, description=description)
|
|
for f in tool_functions
|
|
] + [
|
|
_parse_tool_from_function(
|
|
f, initialize_inputs, tool_type=ToolType.CUSTOM_LLM, name=name, description=description
|
|
)
|
|
for (f, initialize_inputs) in tool_methods
|
|
]
|
|
|
|
|
|
def generate_custom_llm_tools_in_module_as_dict(module, name=None, description=None, **advanced_features):
|
|
tools = generate_custom_llm_tools_in_module(module, name, description)
|
|
return _construct_tool_dict(tools, **advanced_features)
|
|
|
|
|
|
def _construct_tool_dict(tools, **advanced_features):
|
|
return {
|
|
f"{t.module}.{t.class_name}.{t.function}"
|
|
if t.class_name is not None
|
|
else f"{t.module}.{t.function}": asdict_with_advanced_features_without_none(t, **advanced_features)
|
|
for t in tools
|
|
}
|
|
|
|
|
|
class ToolValidationError(UserErrorException):
|
|
"""Base exception raised when failed to validate tool."""
|
|
|
|
def __init__(self, message):
|
|
super().__init__(message, target=ErrorTarget.TOOL)
|
|
|
|
|
|
class PythonParsingError(ToolValidationError):
|
|
pass
|
|
|
|
|
|
class BadFunctionInterface(PythonParsingError):
|
|
pass
|