BaseTool API¶

BaseTool 是所有工具的抽象基类。它定义工具的注册键、参数 schema 解析、上下文注入开关（merge_context）、Prompt 段位声明（prompt_position）以及同步 / 异步双路 run / async_run 框架方法。

设计与使用见工具、运行时上下文注入与 Prompt 段位注入。

Bases: TFNeuralModel, ABC

基础工具是被驱动管理，在Brain工作期间被调用，服务于Robot的封装。

Base Tool is the tool belong to Robot, it works for Brain, and driven by Drive.

Attributes:

Name	Type	Description
`name`	`str`	工具名称，需要全局唯一 \| Tool name, it should be unique in the whole system.
`version`	`Optional[str]`	工具版本 \| Tool version.
`description`	`str`	工具描述 \| Tool description.
`params_schema`	`Optional[dict \| Type[BaseModel]]`	需要注意，如果使用JSON Schema进行参数描述，无法同时实现对Python函数的 args与kwargs 方式的参数调用封装。需要工具内部自行封装实现。 \| The schema of the params of the tool. Notice: If using JSON Schema to describe the parameters, it is impossible to realize the encapsulation of the parameter call of Python functions in args and kwargs modes at the same time. It needs to be encapsulated internally by the tool itself. 如果JSON Schema最外层type为array，在进行函数调用时，将会使用 args 的方式将array中的值全部展开。 \| If the type of the outermost layer of JSON Schema is array, when calling the function, the values in the array will be expanded using args. 如果JSON Schema最外层的type为object，在进行函数调用时，将会使用 kwargs 的方式将object对象展开传入函数参数调用。 \| If the type of the outermost layer of JSON Schema is object, when calling the function, the object will be expanded using kwargs to pass into the function parameter call. 如果使用JSON Schema进行参数描述，根据JSON Type的取值： \| If using JSON Schema to describe the parameters, according to the value of JSON Type: "string" \| "integer" \| "number" \| "boolean": 字符串类型，参数经过合法性校验与解析后，会以位置参数的形式传入 \| String type, after the parameter is validated and parsed, it will be passed in as a positional parameter. "array": 数组类型，参数经过合法性校验与解析后，会以位置参数的形式传入 \| Array type, after the parameter is validated and parsed, it will be passed in as a positional parameter. "object": 对象类型，参数经过合法性校验与解析后，会以关键字参数的形式传入 \| Object type, after the parameter is validated and parsed, it will be passed in as a keyword parameter. "null": 无参数要求 \| No parameter requirements. 如果使用TypeAdapter进行参数描述，根据TypeAdapter(T)其中T的取值有两种情况： 1. T 为普通Python类型约束。比如 list[str], str, int等等。这种情况下，会进行TypeAdapter(T).validate_python方法校验，通过表示全部正常，不通过表示未通过校验 2. T 为Pydantic BaseModel类型。这种情况下：TFRobot会尝试解析Pydantic BaseModel的属性描述信息中是否存在 param_kind 字段（此字段的取值为 inspect._ParameterKind），如果存在，会根据这个字段的取值进行填充。如果不存在，默认设置为 inspect._ParameterKind.KEYWORD_ONLY。填充规则如下： 1. 'positional-only': 会填充到args中 2. 'positional or keyword': 会优先填充到kwargs中，但如果在该字段后出现'variadic positional'类型的字段，将会将此字段转移至args 3. 'variadic positional': 会将此字段展开后，填充到args中 4. 'keyword-only': 会填充到kwargs中 5. 'variadic keyword': 会将此字段展开后，填充到kwargs中

tool_name `cached` `property` ¶

tool_name: str

获取工具的名称。默认使用工具的注册名称。但遇到自定义工具类，可以通过不同配置生成不同的工具实例，而这些工具实例需要同时向LLM提供并工作时，要求工具名称必须不同。此时generate_register_name方法无法满足需求，需要自行实现此方法。

Get the name of the tool. By default, the registered name of the tool is used. However, when encountering custom tool classes, different tool instances can be generated through different configurations. These tool instances need to be provided to the LLM at the same time and require that the tool names must be different when working. In this case, the generate_register_name method cannot meet the requirements, and this method needs to be implemented.

Returns:

Name	Type	Description
`str`	`str`	工具的名称。The name of the tool.

tool_description `cached` `property` ¶

tool_description: str

获取工具描述，一般情况下使用类定义的Description足够了，但少数情况下，开发者会存在自定义工具实例描述的需求（比如MCP工具集）此时需要通过定制此方法实现

Returns:

Name	Type	Description
`str`	`str`	工具描述

params_json_schema `property` ¶

params_json_schema: dict

将params_schema转换为JSON Schema字典对象。

Converts params_schema to a JSON Schema dict.

如果params_schema为None，则返回一个表示空对象的JSON Schema。如果params_schema是字典，则直接转换为JSON字符串。对于 BaseModel类型，调用其schema_json()方法进行转换。

If params_schema is None, returns a JSON Schema representing an empty object. If params_schema is a dictionary, it directly converts to a JSON string. For BaseModel types, calls its model_json_schema() method for conversion.

Returns:

Name	Type	Description
`str`	`dict`	表示参数模式的JSON Schema字符串。A JSON Schema string representing the parameter schema.

validate_attrs `classmethod` ¶

validate_attrs(v: Any) -> Any

Validate the attributes of the tool.

Parameters:

Name	Type	Description	Default
`v`	`Any`	The attributes of the tool.	required

Returns:

Name	Type	Description
`Any`	`Any`	The validated attributes of the tool.

Raises:

Type	Description
`ValueError`	If the tool name is not valid.

Source code in tfrobot/drive/tool/base.py

@model_validator(mode="before")
@classmethod
def validate_attrs(cls, v: Any) -> Any:
    """
    Validate the attributes of the tool.

    Args:
        v (Any): The attributes of the tool.

    Returns:
        Any: The validated attributes of the tool.

    Raises:
        ValueError: If the tool name is not valid.
    """
    if isinstance(v, dict):
        if inspect.isclass(v.get("params_schema")) and issubclass(v["params_schema"], BaseModel):
            v["params_schema"] = TypeAdapter(v.get("params_schema"))
    return v

run ¶

run(
    tool_params: JsonTypes = empty, **kwargs: Any
) -> ToolReturn | Iterator[ToolReturn]

执行工具的逻辑，并验证tool_params，然后返回_run方法的结果。

Executes the tool's logic, validates tool_params, and then returns the result of the _run method.

此方法首先验证tool_params是否遵循指定的参数模式，然后调用_run方法执行具体操作，并返回结果。结果可以是单个ToolReturn类型的实例或这些实例的迭代器。

This method first validates whether tool_params adheres to the specified parameter schema, then calls the _run method to perform specific actions, and returns the results. The result can be either a single instance of type ToolReturn or an iterator of these instances.

在 Python 3.7+ 中，如果一个生成器直接或间接在执行过程中抛出 StopIteration 异常，它会被解释为一个 RuntimeError，以避免与生成器正常结束信号（生成器自然耗尽）混淆。这个改变主要是为了改进和统一异步和同步代码的行为。

背景

在旧版本的 Python 中，生成器可以抛出 StopIteration 来提早结束生成器。然而，这可能导致混淆，因为 StopIteration 通常由 next() 函数自动抛出以表示迭代器已耗尽。为了避免这种情况，在 PEP 479 中建议改变生成器抛出 StopIteration 的处理方式，使其变为 RuntimeError。

所以在此不会再重新抛出Stop异常，而是使用break正常结束迭代器

在此使用while True,配合next()方法与 anext()方法来实现循环迭代主要是为了捕获最终抛出的StopIteration/StopAsyncIteration异常。以满足用户在封装工具时使用 return结尾的写法。因为有时候用户封装迭代器，不仅使用yield，也会使用return语句。而return语句会触发上述异常。

Parameters:

Name	Type	Description	Default
`tool_params`	`JsonTypes`	工具参数，遵循指定的参数模式。Tool parameters, adhering to the specified params schema.	`empty`
`**kwargs`	`Dict[str, Any]`	任意的关键字参数，提供额外的配置选项。Any additional keyword arguments providing extra configuration options.	`{}`

Returns:

Type	Description
`ToolReturn \| Iterator[ToolReturn]`	Union[ToolReturn, Iterator[ToolReturn]]: `_run`方法的执行结果，可能是单个结果或结果的迭代器。The execution result of the `_run` method, which may be a single result or an iterator of results.

Source code in tfrobot/drive/tool/base.py

def run(
    self,
    tool_params: JsonTypes = inspect.Parameter.empty,  # type: ignore
    **kwargs: Any,
) -> ToolReturn | Iterator[ToolReturn]:
    """
    执行工具的逻辑，并验证`tool_params`，然后返回`_run`方法的结果。

    Executes the tool's logic, validates `tool_params`, and then returns the result of the `_run` method.

    此方法首先验证`tool_params`是否遵循指定的参数模式，然后调用`_run`方法执行具体操作，并返回结果。结果可以是单个`ToolReturn`类型的实例或这些实例的迭代器。

    This method first validates whether `tool_params` adheres to the specified parameter schema, then calls the
    `_run` method to perform specific actions, and returns the results. The result can be either a single instance
    of type `ToolReturn` or an iterator of these instances.

    在 Python 3.7+ 中，如果一个生成器直接或间接在执行过程中抛出 StopIteration 异常，它会被解释为一个 RuntimeError，
    以避免与生成器正常结束信号（生成器自然耗尽）混淆。这个改变主要是为了改进和统一异步和同步代码的行为。

    背景

    在旧版本的 Python 中，生成器可以抛出 StopIteration 来提早结束生成器。然而，这可能导致混淆，因为 StopIteration 通常由
    next() 函数自动抛出以表示迭代器已耗尽。为了避免这种情况，在 PEP 479 中建议改变生成器抛出 StopIteration 的处理方式，
    使其变为 RuntimeError。

    所以在此不会再重新抛出Stop异常，而是使用break正常结束迭代器

    在此使用while True,配合next()方法与 __anext__()方法来实现循环迭代主要是为了捕获最终抛出的StopIteration/StopAsyncIteration异常。
    以满足用户在封装工具时使用 return结尾的写法。因为有时候用户封装迭代器，不仅使用yield，也会使用return语句。而return语句会触发上述异常。

    Args:
        tool_params (JsonTypes): 工具参数，遵循指定的参数模式。Tool parameters, adhering to the specified params schema.
        **kwargs (Dict[str, Any]): 任意的关键字参数，提供额外的配置选项。Any additional keyword arguments providing extra
            configuration options.

    Returns:
        Union[ToolReturn, Iterator[ToolReturn]]: `_run`方法的执行结果，可能是单个结果或结果的迭代器。The execution result
            of the `_run` method, which may be a single result or an iterator of results.

    """
    if self._neural:
        kwargs["neural"] = self._neural
    if tool_params != inspect.Parameter.empty:
        # 只有存在tool_params时才进行校验
        if (
            self.params_schema
            and isinstance(self.params_schema, TypeAdapter)
            and not is_validated_pydantic_schema(instance=tool_params, sch=self.params_schema)
        ):
            error_msg = get_pydantic_validation_error_message(instance=tool_params, sch=self.params_schema)
            raise TFExecutionError(f"Tool '{self.tool_name}' parameter validation failed.\n{error_msg}")
        if (
            self.params_schema
            and isinstance(self.params_schema, dict)
            and not is_validated_json_schema(instance=tool_params, sch=self.params_schema)
        ):
            error_msg = get_jsonschema_validation_error_message(instance=tool_params, sch=self.params_schema)
            raise TFExecutionError(f"Tool '{self.tool_name}' parameter validation failed.\n{error_msg}")
        if not self.params_schema and tool_params:
            raise TFExecutionError(f"Tool '{self.tool_name}' takes no parameters but got input.")
    try:
        if self.params_schema:
            tool_args, tool_kwargs = (
                format_parameters_by_json_sch(tool_params, self.params_schema)
                if isinstance(self.params_schema, dict)
                else format_parameters_by_pydantic(tool_params, self.params_schema)
            )
        else:
            tool_args = []
            tool_kwargs = {}
        if self.merge_context:
            # 如果是自行开发的方法，并且严禁接收 **kwargs。需要关闭merge_context，否则会触发函数调用参数错误
            tool_kwargs.update(kwargs)
        observation = self._run(*tool_args, **tool_kwargs)
    except (Exception, KeyboardInterrupt) as e:  # pragma: no cover
        raise e  # pragma: no cover
    if is_attribute_value(observation):
        meta = MessageMeta(tool_name=self.tool_name, arguments=str(tool_params))
        observation = ToolReturn(
            origin=observation,
            result_for_llm=self.__class__.format_output(observation, self) if observation else None,
            msg_meta=meta,
        )
    elif isinstance(observation, Iterator):

        def iterate_with_call_id_update() -> Iterator[ToolReturn]:
            while True:
                try:
                    ob = next(observation)
                    yield wrap_tool_return(
                        ob, self.tool_name, tool_params, lambda obs: self.__class__.format_output(obs, self)
                    )
                except StopIteration as si:
                    final_return = si.value
                    if final_return:
                        final_return = wrap_tool_return(
                            final_return,
                            self.tool_name,
                            tool_params,
                            lambda obs: self.__class__.format_output(obs, self),
                        )
                        yield final_return  # 直接使用 yield 返回最后一个值
                    break  # 正常结束迭代器

        return iterate_with_call_id_update()

    if not isinstance(observation, ToolReturn) and not isinstance(observation, Iterator):
        raise TFProtocolError(f"Invalid tool return type {type(observation)}", protocol="tool_return_schema")
    return observation

async_run `async` ¶

async_run(
    tool_params: JsonTypes = empty, **kwargs: Any
) -> (
    ToolReturn
    | AsyncIterator[ToolReturn]
    | Iterator[ToolReturn]
)

异步验证工具参数并执行工具的异步运行方法。

Asynchronously validates tool parameters and executes the tool's async run method.

首先验证tool_params是否符合params_schema定义。若验证通过，则调用_async_run方法并返回其结果。

First, it validates whether tool_params conform to the params_schema definition. If validation passes, it calls the _async_run method and returns its result.

在 Python 3.7+ 中，如果一个生成器直接或间接在执行过程中抛出 StopIteration 异常，它会被解释为一个 RuntimeError，以避免与生成器正常结束信号（生成器自然耗尽）混淆。这个改变主要是为了改进和统一异步和同步代码的行为。

背景

在旧版本的 Python 中，生成器可以抛出 StopIteration 来提早结束生成器。然而，这可能导致混淆，因为 StopIteration 通常由 next() 函数自动抛出以表示迭代器已耗尽。为了避免这种情况，在 PEP 479 中建议改变生成器抛出 StopIteration 的处理方式，使其变为 RuntimeError。

所以在此不会再重新抛出Stop异常，而是使用break正常结束迭代器

在此使用while True,配合next()方法与 anext()方法来实现循环迭代主要是为了捕获最终抛出的StopIteration/StopAsyncIteration异常。以满足用户在封装工具时使用 return结尾的写法。因为有时候用户封装迭代器，不仅使用yield，也会使用return语句。而return语句会触发上述异常。

Parameters:

Name	Type	Description	Default
`tool_params`	`JsonTypes`	工具参数，遵循指定的参数模式。Tool parameters, adhering to the specified params schema.	`empty`
`**kwargs`	`Dict[str, Any]`	任意的关键字参数，提供额外的配置选项。Any additional keyword arguments providing extra configuration options.	`{}`

Returns:

Type	Description
`ToolReturn \| AsyncIterator[ToolReturn] \| Iterator[ToolReturn]`	ToolReturn \| AsyncIterator[ToolReturn] \| Iterator[ToolReturn]: `_run`方法的执行结果，可能是单个结果或结果的迭代器。 The execution result of the `_run` method, which may be a single result or an iterator of results.

Source code in tfrobot/drive/tool/base.py

async def async_run(
    self,
    tool_params: JsonTypes = inspect.Parameter.empty,  # type: ignore
    **kwargs: Any,
) -> ToolReturn | AsyncIterator[ToolReturn] | Iterator[ToolReturn]:
    """
    异步验证工具参数并执行工具的异步运行方法。

    Asynchronously validates tool parameters and executes the tool's async run method.

    首先验证`tool_params`是否符合`params_schema`定义。若验证通过，则调用`_async_run`方法并返回其结果。

    First, it validates whether `tool_params` conform to the `params_schema` definition. If validation passes,
    it calls the `_async_run` method and returns its result.

    在 Python 3.7+ 中，如果一个生成器直接或间接在执行过程中抛出 StopIteration 异常，它会被解释为一个 RuntimeError，
    以避免与生成器正常结束信号（生成器自然耗尽）混淆。这个改变主要是为了改进和统一异步和同步代码的行为。

    背景

    在旧版本的 Python 中，生成器可以抛出 StopIteration 来提早结束生成器。然而，这可能导致混淆，因为 StopIteration 通常由
    next() 函数自动抛出以表示迭代器已耗尽。为了避免这种情况，在 PEP 479 中建议改变生成器抛出 StopIteration 的处理方式，
    使其变为 RuntimeError。

    所以在此不会再重新抛出Stop异常，而是使用break正常结束迭代器

    在此使用while True,配合next()方法与 __anext__()方法来实现循环迭代主要是为了捕获最终抛出的StopIteration/StopAsyncIteration异常。
    以满足用户在封装工具时使用 return结尾的写法。因为有时候用户封装迭代器，不仅使用yield，也会使用return语句。而return语句会触发上述异常。

    Args:
        tool_params (JsonTypes): 工具参数，遵循指定的参数模式。Tool parameters, adhering to the
            specified params schema.
        **kwargs (Dict[str, Any]): 任意的关键字参数，提供额外的配置选项。Any additional keyword arguments providing extra
            configuration options.

    Returns:
        ToolReturn | AsyncIterator[ToolReturn] | Iterator[ToolReturn]: `_run`方法的执行结果，可能是单个结果或结果的迭代器。
            The execution result of the `_run` method, which may be a single result or an iterator of results.

    """
    if self._neural:
        kwargs["neural"] = self._neural
    if isinstance(self.params_schema, TypeAdapter) and not is_validated_pydantic_schema(
        instance=tool_params, sch=self.params_schema
    ):
        error_msg = get_pydantic_validation_error_message(instance=tool_params, sch=self.params_schema)
        raise TFExecutionError(f"Tool '{self.tool_name}' parameter validation failed.\n{error_msg}")
    if isinstance(self.params_schema, dict) and not is_validated_json_schema(
        instance=tool_params, sch=self.params_schema
    ):
        error_msg = get_jsonschema_validation_error_message(instance=tool_params, sch=self.params_schema)
        raise TFExecutionError(f"Tool '{self.tool_name}' parameter validation failed.\n{error_msg}")
    if not self.params_schema and tool_params:
        raise TFExecutionError(f"Tool '{self.tool_name}' takes no parameters but got input.")
    try:
        if self.params_schema:
            tool_args, tool_kwargs = (
                format_parameters_by_json_sch(tool_params, self.params_schema)
                if isinstance(self.params_schema, dict)
                else format_parameters_by_pydantic(tool_params, self.params_schema)
            )
        else:
            tool_args = []
            tool_kwargs = {}
        if self.merge_context:
            # 如果是自行开发的方法，并且严禁接收 **kwargs。需要关闭merge_context，否则会触发函数调用参数错误
            tool_kwargs.update(kwargs)
        a_run = self._async_run(*tool_args, **tool_kwargs)
        # async 装饰的函数有可能有两种类型，一种是awaitable类型，可以直接使用await拦截。另一种是生成器，需要使用async for 来进行拦截
        if inspect.isawaitable(a_run):
            observation = await a_run
        elif inspect.isasyncgen(a_run):
            observation = a_run
        else:
            raise TFProtocolError(
                f"Invalid async run return type {type(a_run)} for tool {self.tool_name}",
                protocol="tool_return_schema",
            )
    except (Exception, KeyboardInterrupt) as e:  # pragma: no cover
        raise e  # pragma: no cover
    if is_attribute_value(observation):
        meta = MessageMeta(tool_name=self.tool_name, arguments=str(tool_params))
        observation = ToolReturn(
            origin=observation, result_for_llm=self.__class__.format_output(observation, self), msg_meta=meta
        )
    elif isinstance(observation, AsyncIterator):

        async def async_iterate_with_tool_return_construct() -> AsyncIterator[ToolReturn]:
            async for ob in observation:
                if is_attribute_value(ob):
                    msg_meta = MessageMeta(tool_name=self.tool_name, arguments=str(tool_params))
                    yield ToolReturn(
                        origin=ob,
                        result_for_llm=self.__class__.format_output(ob, self) if ob else None,
                        msg_meta=msg_meta,
                    )
                else:
                    yield ob

        return async_iterate_with_tool_return_construct()
    elif isinstance(observation, Iterator):

        def iterate_with_call_id_update() -> Iterator[ToolReturn]:
            while True:
                try:
                    ob = next(observation)
                    yield wrap_tool_return(
                        ob, self.tool_name, tool_params, lambda obs: self.__class__.format_output(obs, self)
                    )
                except StopIteration as si:
                    final_return = si.value
                    if final_return:
                        final_return = wrap_tool_return(
                            final_return,
                            self.tool_name,
                            tool_params,
                            lambda obs: self.__class__.format_output(obs, self),
                        )
                        yield final_return  # 直接使用 yield 返回最后一个值
                    break  # 正常结束迭代器

        return iterate_with_call_id_update()

    if not isinstance(observation, (ToolReturn, AsyncIterator, Iterator)):
        raise TFProtocolError(f"Invalid async tool return type {type(observation)}", protocol="tool_return_schema")
    return observation

run_in_neural ¶

run_in_neural(
    sender: str, **kwargs: Any
) -> ToolReturn | Iterator[ToolReturn]

Run the tool in the neural model.

Parameters:

Name	Type	Description	Default
`sender`	`str`	The sender of the tool.	required
`**kwargs`	`Any`	The arguments for the tool.	`{}`

Returns:

Type	Description
`ToolReturn \| Iterator[ToolReturn]`	ToolReturn \| Iterator[ToolReturn]: The result of the tool.

Source code in tfrobot/drive/tool/base.py

def run_in_neural(self, sender: str, **kwargs: Any) -> ToolReturn | Iterator[ToolReturn]:
    """
    Run the tool in the neural model.

    Args:
        sender(str): The sender of the tool.
        **kwargs(Any): The arguments for the tool.

    Returns:
        ToolReturn | Iterator[ToolReturn]: The result of the tool.
    """
    if sender != self.tool_name:
        raise ValueError(f"Sender {sender} is not the same as the tool {self.tool_name}")
    return self.run(**kwargs)

async_run_in_neural `async` ¶

async_run_in_neural(
    sender: str, **kwargs: Any
) -> (
    ToolReturn
    | AsyncIterator[ToolReturn]
    | Iterator[ToolReturn]
)

Run the tool in the neural model asynchronously.

Parameters:

Name	Type	Description	Default
`sender`	`str`	The sender of the tool.	required
`**kwargs`	`Any`	The arguments for the tool.	`{}`

Returns:

Type	Description
`ToolReturn \| AsyncIterator[ToolReturn] \| Iterator[ToolReturn]`	ToolReturn \| AsyncIterator[ToolReturn] \| Iterator[ToolReturn]: The result of the tool.

Source code in tfrobot/drive/tool/base.py

async def async_run_in_neural(
    self, sender: str, **kwargs: Any
) -> ToolReturn | AsyncIterator[ToolReturn] | Iterator[ToolReturn]:
    """
    Run the tool in the neural model asynchronously.

    Args:
        sender(str): The sender of the tool.
        **kwargs(Any): The arguments for the tool.

    Returns:
        ToolReturn | AsyncIterator[ToolReturn] | Iterator[ToolReturn]: The result of the tool.
    """
    if sender != self.tool_name:
        raise ValueError(f"Sender {sender} is not the same as the tool {self.tool_name}")
    return await self.async_run(**kwargs)

connect_to_neural ¶

connect_to_neural(neural: Neural) -> None

Connect to a neural model.

Notes

Register the async run method with different name (start with "async_") to avoid conflict with the run method.

Parameters:

Name	Type	Description	Default
`neural`	`Neural`	The neural model to connect to.	required

Returns:

Source code in tfrobot/drive/tool/base.py

def connect_to_neural(self, neural: Neural) -> None:
    """
    Connect to a neural model.

    Notes:
        Register the async run method with different name (start with "async_") to avoid conflict with the run
        method.

    Args:
        neural(Neural): The neural model to connect to.

    Returns:

    """
    # self._neural = neural
    setattr(self, "_neural", neural)
    action_name = self.tool_name
    cast(Neural, self._neural).signal.connect(self.run_in_neural, action_name)
    cast(Neural, self._neural).signal.connect(self.async_run_in_neural, f"async_{action_name}")

disconnect_from_neural ¶

disconnect_from_neural(neural: Neural) -> None

Disconnect from a neural model.

Notes

Register the async run method with different name (start with "async_") to avoid conflict with the run method.

Raises:

Type	Description
`ValueError`	If the neural model is not the same as the connected neural model.

Parameters:

Name	Type	Description	Default
`neural`	`Neural`	The neural model to disconnect from.	required

Returns:

Source code in tfrobot/drive/tool/base.py

def disconnect_from_neural(self, neural: "Neural") -> None:
    """
    Disconnect from a neural model.

    Notes:
        Register the async run method with different name (start with "async_") to avoid conflict with the run
        method.

    Raises:
        ValueError: If the neural model is not the same as the connected neural model.

    Args:
        neural(Neural): The neural model to disconnect from.

    Returns:

    """
    if self._neural is not neural:
        raise ValueError("The neural model is not the same as the connected neural model.")
    action_name = generate_register_name(self)
    self._neural.signal.disconnect(self.run_in_neural, action_name)
    self._neural.signal.disconnect(self.async_run_in_neural, f"async_{action_name}")
    # self._neural = None
    setattr(self, "_neural", None)

format_2_prompt ¶

format_2_prompt(
    prompt_ctx: PromptContext,
) -> BasePrompt | None

父类框架方法（子类不应覆写，覆写信号请放在 _format_2_prompt）。

在每次 LLM 调用 prompt 构建阶段，由 LLM 注入器（S2/S3 落地）对 active tools 调用。归一化子类业务点 _format_2_prompt 的六态返回：

子类返回 str → 自动转义字面 { / } 后包成 NormalizePrompt （全仓唯一一处花括号转义点，工具 dev 不必懂 fstring placeholder 规则）
子类返回 BasePrompt 子类实例 → 原样透传
子类返回 "" 或 None → 框架返回 None（注入器跳过该工具贡献）
子类抛异常 → 框架降级为 None + 记 ERROR 日志（保 LLM 调用主链路不挂掉）
子类返回非法类型 → 抛 TFProtocolError（protocol="tool_format_2_prompt"）

关键约束（工具 dev 必读）： - 工具状态存哪由工具自己决定（文件 / DB / 内存 / additional_info / neural 召回均可）。框架仅承诺 prompt_ctx 在调用时已完整构建。 - 此方法在 active tools ∧ 已声明 prompt_position 时每次 LLM 调用都会触发，实现需保证幂等且廉价。昂贵 IO 应在 _run / _async_run 内 prefetch，render 时只读。

Source code in tfrobot/drive/tool/base.py

def format_2_prompt(self, prompt_ctx: "PromptContext") -> "BasePrompt | None":
    """父类框架方法（**子类不应覆写，覆写信号请放在 _format_2_prompt**）。

    在每次 LLM 调用 prompt 构建阶段，由 LLM 注入器（S2/S3 落地）对 active tools 调用。
    归一化子类业务点 _format_2_prompt 的六态返回：

    - 子类返回 ``str``  → 自动转义字面 ``{`` / ``}`` 后包成 NormalizePrompt
      （**全仓唯一一处花括号转义点**，工具 dev 不必懂 fstring placeholder 规则）
    - 子类返回 ``BasePrompt`` 子类实例 → 原样透传
    - 子类返回 ``""`` 或 ``None`` → 框架返回 ``None``（注入器跳过该工具贡献）
    - 子类抛异常 → 框架降级为 ``None`` + 记 ERROR 日志（保 LLM 调用主链路不挂掉）
    - 子类返回非法类型 → 抛 TFProtocolError（protocol="tool_format_2_prompt"）

    关键约束（工具 dev 必读）：
    - 工具状态存哪由工具自己决定（文件 / DB / 内存 / additional_info / neural 召回均可）。
      框架仅承诺 prompt_ctx 在调用时已完整构建。
    - 此方法在 active tools ∧ 已声明 prompt_position 时每次 LLM 调用都会触发，
      实现需保证幂等且廉价。昂贵 IO 应在 _run / _async_run 内 prefetch，render 时只读。
    """
    # 性能短路：未覆写 _format_2_prompt 的工具直接 return None，连 try 都不进
    if self.__class__._format_2_prompt is BaseTool._format_2_prompt:
        return None
    try:
        raw = self._format_2_prompt(prompt_ctx)
    except Exception as e:
        logger.error("Tool %s format_2_prompt failed, falling back to None: %s", self.tool_name, e, exc_info=True)
        return None
    if raw is None or raw == "":
        return None
    if isinstance(raw, str):
        return self._wrap_str_as_normalize_prompt(raw)
    # 延迟 import 避免 tool.base 与 brain.chain.prompt 的模块级循环依赖
    from tfrobot.brain.chain.prompt.base import BasePrompt

    if isinstance(raw, BasePrompt):
        return raw
    raise TFProtocolError(
        f"Tool {self.tool_name}._format_2_prompt() must return str | BasePrompt | None, "
        f"got {type(raw).__name__}",
        protocol="tool_format_2_prompt",
    )

BaseTool API¶

tool_name cached property ¶

tool_description cached property ¶

params_json_schema property ¶

validate_attrs classmethod ¶

run ¶

async_run async ¶

run_in_neural ¶

async_run_in_neural async ¶

connect_to_neural ¶

disconnect_from_neural ¶

format_2_prompt ¶

tool_name `cached` `property` ¶

tool_description `cached` `property` ¶

params_json_schema `property` ¶

validate_attrs `classmethod` ¶

async_run `async` ¶

async_run_in_neural `async` ¶