codekingpro/portable-devtools

1"""Convert functions and runnables to tools."""2 3import inspect4from collections.abc import Callable5from typing import Any, Literal, cast, get_type_hints, overload6 7from pydantic import BaseModel, Field, create_model8 9from langchain_core.callbacks import Callbacks10from langchain_core.runnables import Runnable11from langchain_core.tools.base import ArgsSchema, BaseTool12from langchain_core.tools.simple import Tool13from langchain_core.tools.structured import StructuredTool14 15 16@overload17def tool(18    *,19    description: str | None = None,20    return_direct: bool = False,21    args_schema: ArgsSchema | None = None,22    infer_schema: bool = True,23    response_format: Literal["content", "content_and_artifact"] = "content",24    parse_docstring: bool = False,25    error_on_invalid_docstring: bool = True,26    extras: dict[str, Any] | None = None,27) -> Callable[[Callable | Runnable], BaseTool]: ...28 29 30@overload31def tool(32    name_or_callable: str,33    runnable: Runnable,34    *,35    description: str | None = None,36    return_direct: bool = False,37    args_schema: ArgsSchema | None = None,38    infer_schema: bool = True,39    response_format: Literal["content", "content_and_artifact"] = "content",40    parse_docstring: bool = False,41    error_on_invalid_docstring: bool = True,42    extras: dict[str, Any] | None = None,43) -> BaseTool: ...44 45 46@overload47def tool(48    name_or_callable: Callable,49    *,50    description: str | None = None,51    return_direct: bool = False,52    args_schema: ArgsSchema | None = None,53    infer_schema: bool = True,54    response_format: Literal["content", "content_and_artifact"] = "content",55    parse_docstring: bool = False,56    error_on_invalid_docstring: bool = True,57    extras: dict[str, Any] | None = None,58) -> BaseTool: ...59 60 61@overload62def tool(63    name_or_callable: str,64    *,65    description: str | None = None,66    return_direct: bool = False,67    args_schema: ArgsSchema | None = None,68    infer_schema: bool = True,69    response_format: Literal["content", "content_and_artifact"] = "content",70    parse_docstring: bool = False,71    error_on_invalid_docstring: bool = True,72    extras: dict[str, Any] | None = None,73) -> Callable[[Callable | Runnable], BaseTool]: ...74 75 76def tool(77    name_or_callable: str | Callable | None = None,78    runnable: Runnable | None = None,79    *args: Any,80    description: str | None = None,81    return_direct: bool = False,82    args_schema: ArgsSchema | None = None,83    infer_schema: bool = True,84    response_format: Literal["content", "content_and_artifact"] = "content",85    parse_docstring: bool = False,86    error_on_invalid_docstring: bool = True,87    extras: dict[str, Any] | None = None,88) -> BaseTool | Callable[[Callable | Runnable], BaseTool]:89    """Convert Python functions and `Runnables` to LangChain tools.90 91    Can be used as a decorator with or without arguments to create tools from functions.92 93    Functions can have any signature - the tool will automatically infer input schemas94    unless disabled.95 96    !!! note "Requirements"97 98        - Functions should have type hints for proper schema inference.99        - Functions may accept multiple arguments and return types are flexible;100            outputs will be serialized if needed.101        - When using with `Runnable`, a string name must be provided.102 103    Args:104        name_or_callable: Optional name of the tool or the `Callable` to be105            converted to a tool.106 107            Overrides the function's name.108 109            Must be provided as a positional argument.110        runnable: Optional `Runnable` to convert to a tool.111 112            Must be provided as a positional argument.113        description: Optional description for the tool.114 115            Precedence for the tool description value is as follows:116 117            - This `description` argument (used even if docstring and/or `args_schema`118                are provided)119            - Tool function docstring (used even if `args_schema` is provided)120            - `args_schema` description (used only if `description` and docstring are121                not provided)122        *args: Extra positional arguments.123 124            Must be empty.125        return_direct: Whether to return directly from the tool rather than continuing126            the agent loop.127        args_schema: Optional argument schema for user to specify.128        infer_schema: Whether to infer the schema of the arguments from the function's129            signature.130 131            This also makes the resultant tool accept a dictionary input to its `run()`132            function.133        response_format: The tool response format.134 135            If `'content'`, then the output of the tool is interpreted as the contents136            of a `ToolMessage`.137 138            If `'content_and_artifact'`, then the output is expected to be a two-tuple139            corresponding to the `(content, artifact)` of a `ToolMessage`.140        parse_docstring: If `infer_schema` and `parse_docstring`, will attempt to141            parse parameter descriptions from Google Style function docstrings.142        error_on_invalid_docstring: If `parse_docstring` is provided, configure143            whether to raise `ValueError` on invalid Google Style docstrings.144        extras: Optional provider-specific extra fields for the tool.145 146            Used to pass configuration that doesn't fit into standard tool fields.147            Chat models should process known extras when constructing model payloads.148 149            !!! example150 151                For example, Anthropic-specific fields like `cache_control`,152                `defer_loading`, or `input_examples`.153 154    Raises:155        ValueError: If too many positional arguments are provided (e.g. violating the156            `*args` constraint).157        ValueError: If a `Runnable` is provided without a string name. When using `tool`158            with a `Runnable`, a `str` name must be provided as the `name_or_callable`.159        ValueError: If the first argument is not a string or callable with160            a `__name__` attribute.161        ValueError: If the function does not have a docstring and description162            is not provided and `infer_schema` is `False`.163        ValueError: If `parse_docstring` is `True` and the function has an invalid164            Google-style docstring and `error_on_invalid_docstring` is True.165        ValueError: If a `Runnable` is provided that does not have an object schema.166 167    Returns:168        The tool.169 170    Examples:171        ```python172        @tool173        def search_api(query: str) -> str:174            # Searches the API for the query.175            return176 177 178        @tool("search", return_direct=True)179        def search_api(query: str) -> str:180            # Searches the API for the query.181            return182 183 184        @tool(response_format="content_and_artifact")185        def search_api(query: str) -> tuple[str, dict]:186            return "partial json of results", {"full": "object of results"}187        ```188 189        Parse Google-style docstrings:190 191        ```python192        @tool(parse_docstring=True)193        def foo(bar: str, baz: int) -> str:194            \"\"\"The foo.195 196            Args:197                bar: The bar.198                baz: The baz.199            \"\"\"200            return bar201 202        foo.args_schema.model_json_schema()203        ```204 205        ```python206        {207            "title": "foo",208            "description": "The foo.",209            "type": "object",210            "properties": {211                "bar": {212                    "title": "Bar",213                    "description": "The bar.",214                    "type": "string",215                },216                "baz": {217                    "title": "Baz",218                    "description": "The baz.",219                    "type": "integer",220                },221            },222            "required": ["bar", "baz"],223        }224        ```225 226        Note that parsing by default will raise `ValueError` if the docstring is227        considered invalid. A docstring is considered invalid if it contains arguments228        not in the function signature, or is unable to be parsed into a summary and229        `'Args:'` blocks. Examples below:230 231        ```python232        # No args section233        def invalid_docstring_1(bar: str, baz: int) -> str:234            \"\"\"The foo.\"\"\"235            return bar236 237        # Improper whitespace between summary and args section238        def invalid_docstring_2(bar: str, baz: int) -> str:239            \"\"\"The foo.240            Args:241                bar: The bar.242                baz: The baz.243            \"\"\"244            return bar245 246        # Documented args absent from function signature247        def invalid_docstring_3(bar: str, baz: int) -> str:248            \"\"\"The foo.249 250            Args:251                banana: The bar.252                monkey: The baz.253            \"\"\"254            return bar255 256        ```257    """  # noqa: D214, D410, D411  # We're intentionally showing bad formatting in examples258 259    def _create_tool_factory(260        tool_name: str,261    ) -> Callable[[Callable | Runnable], BaseTool]:262        """Create a decorator that takes a callable and returns a tool.263 264        Args:265            tool_name: The name that will be assigned to the tool.266 267        Returns:268            A function that takes a callable or `Runnable` and returns a tool.269        """270 271        def _tool_factory(dec_func: Callable | Runnable) -> BaseTool:272            tool_description = description273            if isinstance(dec_func, Runnable):274                runnable = dec_func275 276                if runnable.input_schema.model_json_schema().get("type") != "object":277                    msg = "Runnable must have an object schema."278                    raise ValueError(msg)279 280                async def ainvoke_wrapper(281                    callbacks: Callbacks | None = None, **kwargs: Any282                ) -> Any:283                    return await runnable.ainvoke(kwargs, {"callbacks": callbacks})284 285                def invoke_wrapper(286                    callbacks: Callbacks | None = None, **kwargs: Any287                ) -> Any:288                    return runnable.invoke(kwargs, {"callbacks": callbacks})289 290                coroutine = ainvoke_wrapper291                func = invoke_wrapper292                schema: ArgsSchema | None = runnable.input_schema293                tool_description = description or repr(runnable)294            elif inspect.iscoroutinefunction(dec_func):295                coroutine = dec_func296                func = None297                schema = args_schema298            else:299                coroutine = None300                func = dec_func301                schema = args_schema302 303            if infer_schema or args_schema is not None:304                return StructuredTool.from_function(305                    func,306                    coroutine,307                    name=tool_name,308                    description=tool_description,309                    return_direct=return_direct,310                    args_schema=schema,311                    infer_schema=infer_schema,312                    response_format=response_format,313                    parse_docstring=parse_docstring,314                    error_on_invalid_docstring=error_on_invalid_docstring,315                    extras=extras,316                )317            # If someone doesn't want a schema applied, we must treat it as318            # a simple string->string function319            if dec_func.__doc__ is None:320                msg = (321                    "Function must have a docstring if "322                    "description not provided and infer_schema is False."323                )324                raise ValueError(msg)325            return Tool(326                name=tool_name,327                func=func,328                description=f"{tool_name} tool",329                return_direct=return_direct,330                coroutine=coroutine,331                response_format=response_format,332                extras=extras,333            )334 335        return _tool_factory336 337    if len(args) != 0:338        # Triggered if a user attempts to use positional arguments that339        # do not exist in the function signature340        # e.g., @tool("name", runnable, "extra_arg")341        # Here, "extra_arg" is not a valid argument342        msg = "Too many arguments for tool decorator. A decorator "343        raise ValueError(msg)344 345    if runnable is not None:346        # tool is used as a function347        # for instance tool_from_runnable = tool("name", runnable)348        if not name_or_callable:349            msg = "Runnable without name for tool constructor"350            raise ValueError(msg)351        if not isinstance(name_or_callable, str):352            msg = "Name must be a string for tool constructor"353            raise ValueError(msg)354        return _create_tool_factory(name_or_callable)(runnable)355    if name_or_callable is not None:356        if callable(name_or_callable) and hasattr(name_or_callable, "__name__"):357            # Used as a decorator without parameters358            # @tool359            # def my_tool():360            #    pass361            return _create_tool_factory(name_or_callable.__name__)(name_or_callable)362        if isinstance(name_or_callable, str):363            # Used with a new name for the tool364            # @tool("search")365            # def my_tool():366            #    pass367            #368            # or369            #370            # @tool("search", parse_docstring=True)371            # def my_tool():372            #    pass373            return _create_tool_factory(name_or_callable)374        msg = (375            f"The first argument must be a string or a callable with a __name__ "376            f"for tool decorator. Got {type(name_or_callable)}"377        )378        raise ValueError(msg)379 380    # Tool is used as a decorator with parameters specified381    # @tool(parse_docstring=True)382    # def my_tool():383    #    pass384    def _partial(func: Callable | Runnable) -> BaseTool:385        """Partial function that takes a `Callable` and returns a tool."""386        name_ = func.get_name() if isinstance(func, Runnable) else func.__name__387        tool_factory = _create_tool_factory(name_)388        return tool_factory(func)389 390    return _partial391 392 393def _get_description_from_runnable(runnable: Runnable) -> str:394    """Generate a placeholder description of a `Runnable`."""395    input_schema = runnable.input_schema.model_json_schema()396    return f"Takes {input_schema}."397 398 399def _get_schema_from_runnable_and_arg_types(400    runnable: Runnable,401    name: str,402    arg_types: dict[str, type] | None = None,403) -> type[BaseModel]:404    """Infer `args_schema` for tool."""405    if arg_types is None:406        try:407            arg_types = get_type_hints(runnable.InputType)408        except TypeError as e:409            msg = (410                "Tool input must be str or dict. If dict, dict arguments must be "411                "typed. Either annotate types (e.g., with TypedDict) or pass "412                f"arg_types into `.as_tool` to specify. {e}"413            )414            raise TypeError(msg) from e415    fields = {key: (key_type, Field(...)) for key, key_type in arg_types.items()}416    return cast("type[BaseModel]", create_model(name, **fields))  # type: ignore[call-overload]417 418 419def convert_runnable_to_tool(420    runnable: Runnable,421    args_schema: type[BaseModel] | None = None,422    *,423    name: str | None = None,424    description: str | None = None,425    arg_types: dict[str, type] | None = None,426) -> BaseTool:427    """Convert a `Runnable` into a `BaseTool`.428 429    Args:430        runnable: The `Runnable` to convert.431        args_schema: The schema for the tool's input arguments.432        name: The name of the tool.433        description: The description of the tool.434        arg_types: The types of the arguments.435 436    Returns:437        The tool.438    """439    if args_schema:440        runnable = runnable.with_types(input_type=args_schema)441    description = description or _get_description_from_runnable(runnable)442    name = name or runnable.get_name()443 444    schema = runnable.input_schema.model_json_schema()445    if schema.get("type") == "string":446        return Tool(447            name=name,448            func=runnable.invoke,449            coroutine=runnable.ainvoke,450            description=description,451        )452 453    async def ainvoke_wrapper(callbacks: Callbacks | None = None, **kwargs: Any) -> Any:454        return await runnable.ainvoke(kwargs, config={"callbacks": callbacks})455 456    def invoke_wrapper(callbacks: Callbacks | None = None, **kwargs: Any) -> Any:457        return runnable.invoke(kwargs, config={"callbacks": callbacks})458 459    if (460        arg_types is None461        and schema.get("type") == "object"462        and schema.get("properties")463    ):464        args_schema = runnable.input_schema465    else:466        args_schema = _get_schema_from_runnable_and_arg_types(467            runnable, name, arg_types=arg_types468        )469 470    return StructuredTool.from_function(471        name=name,472        func=invoke_wrapper,473        coroutine=ainvoke_wrapper,474        description=description,475        args_schema=args_schema,476    )477