flowchart TB
    subgraph UserCode["用户代码层"]
        USER[用户应用<br/>定义工具函数]
    end

    subgraph ToolCreation["工具创建层"]
        DECORATOR[@tool装饰器]
        FROM_FUNC[from_function方法]
        SUBCLASS[继承BaseTool]
        SCHEMA_GEN[create_schema_from_function<br/>Schema生成器]
    end

    subgraph ToolCore["工具核心层"]
        BASE_TOOL[BaseTool<br/>抽象基类]
        TOOL[Tool<br/>简单工具]
        STRUCTURED[StructuredTool<br/>结构化工具]
        RETRIEVER_TOOL[RetrieverTool<br/>检索工具]
    end

    subgraph Integration["模型集成层"]
        BIND_TOOLS[bind_tools<br/>绑定工具到模型]
        TOOL_SCHEMA[tool_call_schema<br/>生成模型schema]
        FORMAT[convert_to_openai_tool<br/>格式转换]
    end

    subgraph LLM["语言模型层"]
        CHAT_MODEL[ChatModel<br/>聊天模型]
        LLM_INVOKE[invoke/ainvoke<br/>模型调用]
    end

    subgraph Execution["工具执行层"]
        INVOKE[invoke/ainvoke<br/>工具调用入口]
        RUN[run/arun<br/>执行管理]
        PARSE_INPUT[_parse_input<br/>输入解析]
        TO_ARGS[_to_args_and_kwargs<br/>参数转换]
        RUN_IMPL[_run/_arun<br/>实际执行]
    end

    subgraph Callback["回调追踪层"]
        CB_START[on_tool_start]
        CB_END[on_tool_end]
        CB_ERROR[on_tool_error]
    end

    subgraph Output["输出处理层"]
        FORMAT_OUTPUT[_format_output<br/>格式化输出]
        TOOL_MSG[ToolMessage<br/>工具消息]
    end

    USER --> DECORATOR
    USER --> FROM_FUNC
    USER --> SUBCLASS

    DECORATOR --> SCHEMA_GEN
    FROM_FUNC --> SCHEMA_GEN
    SCHEMA_GEN --> STRUCTURED
    SCHEMA_GEN --> TOOL
    SUBCLASS --> BASE_TOOL

    BASE_TOOL <|-- TOOL
    BASE_TOOL <|-- STRUCTURED
    BASE_TOOL <|-- RETRIEVER_TOOL

    TOOL --> BIND_TOOLS
    STRUCTURED --> BIND_TOOLS
    RETRIEVER_TOOL --> BIND_TOOLS

    BIND_TOOLS --> TOOL_SCHEMA
    TOOL_SCHEMA --> FORMAT
    FORMAT --> CHAT_MODEL

    CHAT_MODEL --> LLM_INVOKE
    LLM_INVOKE -.返回ToolCall.-> INVOKE

    INVOKE --> RUN
    RUN --> CB_START
    CB_START --> PARSE_INPUT
    PARSE_INPUT --> TO_ARGS
    TO_ARGS --> RUN_IMPL
    RUN_IMPL -.成功.-> CB_END
    RUN_IMPL -.失败.-> CB_ERROR
    CB_END --> FORMAT_OUTPUT
    FORMAT_OUTPUT --> TOOL_MSG

    style BASE_TOOL fill:#e1f5ff,stroke:#0277bd,stroke-width:3px
    style DECORATOR fill:#fff4e1,stroke:#f57c00,stroke-width:2px
    style STRUCTURED fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
    style BIND_TOOLS fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px
    style RUN_IMPL fill:#ffebee,stroke:#c62828,stroke-width:2px

# 策略1: 返回错误信息(不中断)
@tool(handle_tool_error=True)
def risky_tool(input: str) -> str:
    if bad_input:
        raise ToolException("Invalid")
    return result

# 策略2: 自定义错误信息
@tool(handle_tool_error="Tool failed, using default value")
def with_default(input: str) -> str:
    ...

# 策略3: 自定义错误处理
@tool(handle_tool_error=lambda e: f"Error: {e}, retrying...")
def with_handler(input: str) -> str:
    ...

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Decorator as @tool装饰器
    participant Schema as create_schema_from_function
    participant StructTool as StructuredTool
    participant Model as ChatModel
    participant BindTools as bind_tools
    participant ToolSchema as tool_call_schema
    participant Converter as convert_to_openai_tool
    participant Invoke as invoke
    participant Run as run
    participant Callback as CallbackManager
    participant RunImpl as _run实现

    Note over User,RunImpl: 阶段1: 工具定义与创建

    User->>Decorator: @tool<br/>def search(query: str): ...
    activate Decorator
    Decorator->>Decorator: 提取函数名称和docstring
    Decorator->>Schema: create_schema_from_function<br/>(name, func, parse_docstring=True)
    activate Schema
    Schema->>Schema: inspect.signature(func)<br/>获取函数签名
    Schema->>Schema: get_type_hints(func)<br/>提取类型注解
    Schema->>Schema: _parse_google_docstring<br/>解析参数描述
    Schema->>Schema: _create_subset_model<br/>构建Pydantic模型
    Schema-->>Decorator: ArgsSchema<br/>(Pydantic模型)
    deactivate Schema

    Decorator->>StructTool: StructuredTool.from_function<br/>(func, name, description, args_schema)
    activate StructTool
    StructTool->>StructTool: 验证参数完整性
    StructTool-->>Decorator: StructuredTool实例
    deactivate StructTool
    Decorator-->>User: search工具对象
    deactivate Decorator

    Note over User,RunImpl: 阶段2: 工具绑定到模型

    User->>Model: model.bind_tools([search, calculator])
    activate Model
    Model->>BindTools: bind_tools(tools)
    activate BindTools

    loop 每个工具
        BindTools->>ToolSchema: tool.tool_call_schema<br/>获取工具schema
        activate ToolSchema
        ToolSchema->>ToolSchema: 过滤注入参数<br/>(InjectedToolCallId等)
        ToolSchema-->>BindTools: 过滤后的schema
        deactivate ToolSchema

        BindTools->>Converter: convert_to_openai_tool(schema)
        activate Converter
        Converter->>Converter: 转换为OpenAI函数格式<br/>{"name":..., "parameters":...}
        Converter-->>BindTools: OpenAI工具定义
        deactivate Converter
    end

    BindTools-->>Model: tools参数<br/>(模型特定格式)
    deactivate BindTools
    Model-->>User: 绑定工具的模型
    deactivate Model

    Note over User,RunImpl: 阶段3: 模型调用与工具选择

    User->>Model: model.invoke("搜索Python教程")
    activate Model
    Model->>Model: 将工具注入prompt<br/>作为函数定义
    Model->>Model: 调用底层LLM API<br/>(OpenAI/Anthropic等)
    Model-->>User: AIMessage(tool_calls=[<br/>  {"name":"search",<br/>   "args":{"query":"Python教程"},<br/>   "id":"call_123"}<br/>])
    deactivate Model

    Note over User,RunImpl: 阶段4: 工具执行

    User->>Invoke: search.invoke(tool_call)
    activate Invoke
    Invoke->>Invoke: _prep_run_args<br/>解析ToolCall
    Invoke->>Run: run(tool_input, tool_call_id)
    activate Run

    Run->>Callback: on_tool_start<br/>(name, input)
    activate Callback
    Callback->>Callback: 记录开始时间
    Callback->>Callback: 触发用户回调
    Callback-->>Run: RunManager
    deactivate Callback

    Run->>Run: _parse_input<br/>验证参数类型
    Run->>Run: _to_args_and_kwargs<br/>转换为函数参数
    Run->>RunImpl: _run(query="Python教程",<br/>     run_manager=...)
    activate RunImpl
    RunImpl->>RunImpl: 执行实际业务逻辑<br/>(API调用/数据库查询等)

    alt 执行成功
        RunImpl-->>Run: 工具结果
        Run->>Run: _format_output<br/>包装为ToolMessage
        Run->>Callback: on_tool_end(output)
        activate Callback
        Callback->>Callback: 记录结束时间
        Callback->>Callback: 计算耗时
        deactivate Callback
        Run-->>Invoke: ToolMessage(content=...,<br/>              tool_call_id="call_123")
    else 工具异常
        RunImpl-->>Run: ToolException
        Run->>Run: _handle_tool_error<br/>根据handle_tool_error策略处理
        alt handle_tool_error=True
            Run->>Callback: on_tool_error(error)
            Run-->>Invoke: ToolMessage(content="Error:...",<br/>              status="error")
        else handle_tool_error=False
            Run->>Callback: on_tool_error(error)
            Run-->>Invoke: 抛出异常
        end
    end
    deactivate RunImpl
    deactivate Run
    Invoke-->>User: ToolMessage
    deactivate Invoke

    Note over User,RunImpl: 阶段5: 结果返回模型

    User->>Model: model.invoke([<br/>  HumanMessage("搜索Python教程"),<br/>  AIMessage(tool_calls=[...]),<br/>  ToolMessage(...)<br/>])
    activate Model
    Model->>Model: 基于工具结果生成最终答案
    Model-->>User: AIMessage(content="根据搜索结果...")
    deactivate Model

# 策略1: 返回错误信息(不中断Agent)
handle_tool_error=True
# -> ToolMessage(content="Error: ...", status="error")

# 策略2: 自定义错误信息
handle_tool_error="工具执行失败,请重试"
# -> ToolMessage(content="工具执行失败,请重试", status="error")

# 策略3: 自定义处理函数
handle_tool_error=lambda e: f"错误: {e}, 使用默认值"
# -> ToolMessage(content="错误: ..., 使用默认值", status="error")

# 策略4: 抛出异常(中断Agent)
handle_tool_error=False  # 默认
# -> 向上抛出ToolException,Agent停止执行

# response_format="content"
ToolMessage(
    content="搜索结果...",
    tool_call_id="call_123",
    name="search",
    status="success"
)

# response_format="content_and_artifact"
ToolMessage(
    content="搜索结果摘要",
    artifact={"results": [...], "total": 10},  # 原始数据
    tool_call_id="call_123",
    name="search",
    status="success"
)

用户 -> 模型 -> 工具1 -> 模型 -> 工具2 -> 模型 -> 最终答案

classDiagram
    class BaseTool {
        <<abstract>>
        +name: str
        +description: str
        +args_schema: Type[BaseModel]|None
        +return_direct: bool
        +handle_tool_error: bool|str|Callable
        +response_format: str
        +invoke(input) Any
        +_run(...)* Any
        +_arun(...) Any
    }

    class Tool {
        +func: Callable
        +coroutine: Callable|None
        +from_function(...) Tool
    }

    class StructuredTool {
        +func: Callable
        +coroutine: Callable|None
        +from_function(...) StructuredTool
    }

    class ToolCall {
        +name: str
        +args: dict
        +id: str
        +type: str
    }

    class ToolException {
        <<Exception>>
    }

    BaseTool <|-- Tool
    BaseTool <|-- StructuredTool
    BaseTool ..> ToolCall : accepts
    BaseTool ..> ToolException : raises

sequenceDiagram
    autonumber
    participant UserFunc as 用户函数
    participant Tool as tool装饰器
    participant Factory as _create_tool_factory
    participant CreateSchema as create_schema_from_function
    participant ParseDoc as _parse_google_docstring
    participant StructTool as StructuredTool.from_function

    Note over UserFunc,StructTool: 装饰时执行(模块加载阶段)

    UserFunc->>Tool: @tool<br/>def search(query: str): ...
    activate Tool
    Tool->>Tool: 检测调用模式<br/>(直接装饰/带参数/指定名称)
    Tool->>Factory: _create_tool_factory("search")
    activate Factory
    Factory->>Factory: 返回_tool_factory闭包
    deactivate Factory

    Tool->>Factory: _tool_factory(user_func)
    activate Factory
    Factory->>Factory: 提取函数名和docstring
    Factory->>Factory: inspect.iscoroutinefunction(func)<br/>检测异步函数

    alt parse_docstring=True
        Factory->>CreateSchema: create_schema_from_function<br/>(name, func, parse_docstring=True)
        activate CreateSchema
        CreateSchema->>CreateSchema: inspect.signature(func)
        CreateSchema->>CreateSchema: get_type_hints(func)
        CreateSchema->>ParseDoc: _parse_google_docstring(docstring)
        activate ParseDoc
        ParseDoc->>ParseDoc: 分割summary和Args部分
        ParseDoc->>ParseDoc: 解析每个参数描述
        ParseDoc-->>CreateSchema: (description, arg_descriptions)
        deactivate ParseDoc
        CreateSchema->>CreateSchema: _create_subset_model<br/>构建Pydantic模型
        CreateSchema-->>Factory: args_schema
        deactivate CreateSchema
    end

    Factory->>StructTool: StructuredTool.from_function<br/>(func, name, description, args_schema)
    activate StructTool
    StructTool->>StructTool: 验证参数完整性
    StructTool->>StructTool: 创建工具实例
    StructTool-->>Factory: StructuredTool实例
    deactivate StructTool
    Factory-->>Tool: BaseTool实例
    deactivate Factory
    Tool-->>UserFunc: 返回工具对象
    deactivate Tool

# 方式1: 直接装饰
@tool
def search(query: str) -> str:
    \"\"\"搜索工具\"\"\"
    return api_search(query)

# 方式2: 指定名称
@tool("custom_search")
def my_search(query: str) -> str:
    return api_search(query)

# 方式3: 带参数
@tool(parse_docstring=True, return_direct=True)
def calculator(expr: str) -> float:
    \"\"\"计算表达式

    Args:
        expr: 数学表达式
    \"\"\"
    return eval(expr)

def create_schema_from_function(model_name: str, func: Callable, *, parse_docstring: bool = False) -> type[BaseModel]:
    # 1) 获取函数签名
    sig = inspect.signature(func)

    # 2) 使用Pydantic的validate_arguments生成模型
    validated = validate_arguments(func, config=_SchemaConfig)
    inferred_model = validated.model

    # 3) 解析docstring提取参数描述
    if parse_docstring:
        description, arg_descriptions = _parse_python_function_docstring(func, annotations)

    # 4) 过滤特殊参数(run_manager, callbacks等)
    filter_args_ = list(FILTERED_ARGS) + [注入参数]

    # 5) 创建子模型(只包含用户参数)
    return _create_subset_model(model_name, inferred_model, valid_properties, descriptions=arg_descriptions)

sequenceDiagram
    autonumber
    participant User as 用户/Agent
    participant Invoke as tool.invoke
    participant Run as tool.run
    participant CBStart as on_tool_start
    participant ParseInput as _parse_input
    participant RunImpl as _run实现
    participant FormatOutput as _format_output
    participant CBEnd as on_tool_end

    User->>Invoke: tool.invoke(tool_call, config)
    activate Invoke
    Invoke->>Invoke: _prep_run_args<br/>解析ToolCall
    Invoke->>Run: run(tool_input, tool_call_id)
    activate Run

    Run->>Run: CallbackManager.configure<br/>配置回调
    Run->>CBStart: on_tool_start(name, input)
    activate CBStart
    CBStart->>CBStart: 记录开始时间
    CBStart->>CBStart: 生成run_id
    CBStart-->>Run: run_manager
    deactivate CBStart

    Run->>ParseInput: _parse_input(tool_input, tool_call_id)
    activate ParseInput
    ParseInput->>ParseInput: Pydantic验证参数
    ParseInput->>ParseInput: 注入tool_call_id(如需要)
    ParseInput-->>Run: 验证后的参数
    deactivate ParseInput

    Run->>Run: _to_args_and_kwargs<br/>转换为函数参数
    Run->>Run: 注入run_manager<br/>注入config

    Run->>RunImpl: _run(*args, **kwargs)
    activate RunImpl
    RunImpl->>RunImpl: 执行业务逻辑

    alt 执行成功
        RunImpl-->>Run: result
        Run->>FormatOutput: _format_output(content, artifact, tool_call_id)
        activate FormatOutput
        FormatOutput->>FormatOutput: 创建ToolMessage
        FormatOutput-->>Run: ToolMessage
        deactivate FormatOutput
        Run->>CBEnd: on_tool_end(output)
        activate CBEnd
        CBEnd->>CBEnd: 记录结束时间
        CBEnd->>CBEnd: 计算耗时
        deactivate CBEnd
        Run-->>Invoke: ToolMessage
    else 工具异常
        RunImpl-->>Run: ToolException
        Run->>Run: _handle_tool_error
        alt handle_tool_error=True
            Run->>CBEnd: on_tool_error
            Run-->>Invoke: ToolMessage(status="error")
        else handle_tool_error=False
            Run->>CBEnd: on_tool_error
            Run-->>Invoke: 抛出异常
        end
    end
    deactivate RunImpl
    deactivate Run
    Invoke-->>User: 工具执行结果
    deactivate Invoke

def run(self, tool_input, callbacks, tags, metadata, config, tool_call_id, **kwargs):
    # 1) 配置回调管理器
    callback_manager = CallbackManager.configure(callbacks, self.callbacks, self.verbose, tags, self.tags, metadata, self.metadata)

    # 2) 启动追踪
    run_manager = callback_manager.on_tool_start(
        {"name": self.name, "description": self.description},
        tool_input,
        run_name=run_name,
        run_id=run_id,
    )

    # 3) 执行工具
    try:
        # 3.1) 转换参数
        tool_args, tool_kwargs = self._to_args_and_kwargs(tool_input, tool_call_id)

        # 3.2) 注入上下文
        if signature(self._run).parameters.get("run_manager"):
            tool_kwargs["run_manager"] = run_manager
        if config_param := _get_runnable_config_param(self._run):
            tool_kwargs[config_param] = config

        # 3.3) 执行
        response = self._run(*tool_args, **tool_kwargs)

        # 3.4) 处理响应格式
        if self.response_format == "content_and_artifact":
            content, artifact = response
        else:
            content = response

    except ValidationError as e:
        # 参数验证错误
        if self.handle_validation_error:
            content = _handle_validation_error(e, self.handle_validation_error)
            status = "error"
        else:
            run_manager.on_tool_error(e)
            raise

    except ToolException as e:
        # 工具执行错误
        if self.handle_tool_error:
            content = _handle_tool_error(e, self.handle_tool_error)
            status = "error"
        else:
            run_manager.on_tool_error(e)
            raise

    # 4) 格式化输出
    output = _format_output(content, artifact, tool_call_id, self.name, status)

    # 5) 结束追踪
    run_manager.on_tool_end(output)

    return output

def _parse_input(self, tool_input: str | dict, tool_call_id: str | None) -> str | dict:
    if isinstance(tool_input, str):
        # 字符串输入:单参数工具
        if self.args_schema:
            key_ = next(iter(get_fields(self.args_schema).keys()))
            self.args_schema.model_validate({key_: tool_input})
        return tool_input

    # 字典输入:Pydantic验证
    if self.args_schema:
        # 注入tool_call_id(如果工具需要)
        for k, v in get_all_basemodel_annotations(self.args_schema).items():
            if _is_injected_arg_type(v, injected_type=InjectedToolCallId):
                if tool_call_id is None:
                    raise ValueError("缺少tool_call_id")
                tool_input[k] = tool_call_id

        # 验证
        result = self.args_schema.model_validate(tool_input)

        # 只返回用户提供的参数
        return {k: getattr(result, k) for k in tool_input}

    return tool_input

def _format_output(content, artifact, tool_call_id, name, status):
    # 如果已经是ToolMessage或无tool_call_id,直接返回
    if isinstance(content, ToolOutputMixin) or tool_call_id is None:
        return content

    # 确保content是字符串或列表格式
    if not _is_message_content_type(content):
        content = json.dumps(content)

    # 创建ToolMessage
    return ToolMessage(content, artifact=artifact, tool_call_id=tool_call_id, name=name, status=status)

tool = create_my_tool()

# 线程安全
with ThreadPoolExecutor() as executor:
    results = list(executor.map(tool.invoke, inputs))

# 协程并发
results = await asyncio.gather(*[tool.ainvoke(inp) for inp in inputs])

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Model as ChatModel
    participant BindTools as bind_tools
    participant ToolSchema as tool_call_schema
    participant Converter as convert_to_openai_tool

    User->>Model: model.bind_tools([search, calculator])
    activate Model
    Model->>BindTools: bind_tools(tools)
    activate BindTools

    loop 每个工具
        BindTools->>ToolSchema: tool.tool_call_schema
        activate ToolSchema
        ToolSchema->>ToolSchema: 获取args_schema
        ToolSchema->>ToolSchema: 过滤注入参数<br/>(InjectedToolCallId, run_manager等)
        ToolSchema->>ToolSchema: _create_subset_model<br/>创建干净的schema
        ToolSchema-->>BindTools: 过滤后的schema
        deactivate ToolSchema

        BindTools->>Converter: convert_to_openai_tool(schema)
        activate Converter
        Converter->>Converter: 转换为OpenAI函数格式<br/>{"name":..., "parameters":...}
        Converter-->>BindTools: OpenAI工具定义
        deactivate Converter
    end

    BindTools->>BindTools: 生成tools列表
    BindTools-->>Model: 返回tools参数
    deactivate BindTools

    Model->>Model: 创建新模型实例<br/>self.bind(tools=tools)
    Model-->>User: 绑定工具的模型
    deactivate Model

@property
def tool_call_schema(self) -> ArgsSchema:
    """获取用于模型调用的schema(过滤注入参数)"""
    if isinstance(self.args_schema, dict):
        if self.description:
            return {**self.args_schema, "description": self.description}
        return self.args_schema

    # 获取完整schema
    full_schema = self.get_input_schema()

    # 过滤注入参数
    fields = []
    for name, type_ in get_all_basemodel_annotations(full_schema).items():
        if not _is_injected_arg_type(type_):
            fields.append(name)

    # 创建子模型
    return _create_subset_model(self.name, full_schema, fields, fn_description=self.description)

def convert_to_openai_tool(tool: BaseTool | type[BaseModel] | dict) -> dict:
    \"\"\"转换为OpenAI函数格式\"\"\"
    if isinstance(tool, BaseTool):
        schema = tool.tool_call_schema
        name = tool.name
        description = tool.description
    else:
        # 处理Pydantic模型或dict
        ...

    # 转换为OpenAI格式
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": schema.model_json_schema()
        }
    }

{
  "type": "function",
  "function": {
    "name": "search",
    "description": "搜索互联网信息",
    "parameters": {
      "type": "object",
      "properties": {
        "query": {
          "type": "string",
          "description": "搜索关键词"
        },
        "limit": {
          "type": "integer",
          "description": "结果数量",
          "default": 10
        }
      },
      "required": ["query"]
    }
  }
}

def bind_tools(self, tools: Sequence[dict | type | Callable | BaseTool], *, tool_choice: str | None = None, **kwargs) -> Runnable:
    \"\"\"绑定工具到模型\"\"\"
    # 转换所有工具为统一格式
    formatted_tools = [convert_to_openai_tool(tool) for tool in tools]

    # 调用self.bind绑定参数
    return self.bind(tools=formatted_tools, tool_choice=tool_choice, **kwargs)

{
  "type": "function",
  "function": {"name": "...", "description": "...", "parameters": {...}}
}

{
  "name": "...",
  "description": "...",
  "input_schema": {...}
}

{
  "name": "...",
  "description": "...",
  "parameters": {...}
}

name_or_callable: str | Callable | None  # 工具名或函数
description: str | None  # 工具描述
return_direct: bool  # 是否直接返回
args_schema: ArgsSchema | None  # 自定义参数schema
infer_schema: bool  # 是否推导schema
response_format: str  # 响应格式
parse_docstring: bool  # 是否解析docstring

BaseTool | Callable[[Callable], BaseTool]
# 返回工具实例或装饰器函数

def tool(
    name_or_callable: str | Callable | None = None,
    *,
    description: str | None = None,
    return_direct: bool = False,
    args_schema: ArgsSchema | None = None,
    infer_schema: bool = True,
    response_format: Literal["content", "content_and_artifact"] = "content",
    parse_docstring: bool = False,
) -> BaseTool | Callable[[Callable], BaseTool]:
    """将函数转换为工具

    Args:
        name_or_callable: 工具名称或函数
        description: 工具描述
        ...其他参数

    Returns:
        工具实例或装饰器函数
    """
    def _create_tool_factory(tool_name: str) -> Callable:
        """创建工具工厂函数"""
        def _tool_factory(dec_func: Callable) -> BaseTool:
            # 1) 确定工具描述
            tool_description = description
            if tool_description is None:
                if dec_func.__doc__:
                    tool_description = dec_func.__doc__.strip()
                elif args_schema:
                    tool_description = args_schema.__doc__

            # 2) 分离同步和异步函数
            if inspect.iscoroutinefunction(dec_func):
                func = None
                coroutine = dec_func
            else:
                func = dec_func
                coroutine = None

            # 3) 处理参数schema
            schema = args_schema
            if parse_docstring:
                # 从docstring解析参数描述
                schema = create_schema_from_function(
                    tool_name,
                    dec_func,
                    parse_docstring=True,
                )

            # 4) 创建StructuredTool或Tool
            if infer_schema or args_schema is not None:
                return StructuredTool.from_function(
                    func,
                    coroutine,
                    name=tool_name,
                    description=tool_description,
                    return_direct=return_direct,
                    args_schema=schema,
                    infer_schema=infer_schema,
                    response_format=response_format,
                )
            else:
                # 简单的字符串工具
                return Tool(
                    name=tool_name,
                    func=func,
                    description=tool_description,
                    return_direct=return_direct,
                    coroutine=coroutine,
                    response_format=response_format,
                )

        return _tool_factory

    # 5) 处理不同的调用方式
    if name_or_callable is not None:
        if callable(name_or_callable) and hasattr(name_or_callable, "__name__"):
            # @tool 直接装饰
            return _create_tool_factory(name_or_callable.__name__)(name_or_callable)
        elif isinstance(name_or_callable, str):
            # @tool("custom_name") 指定名称
            return _create_tool_factory(name_or_callable)

    # @tool(parse_docstring=True) 带参数装饰
    def _partial(func: Callable) -> BaseTool:
        name = func.__name__
        tool_factory = _create_tool_factory(name)
        return tool_factory(func)

    return _partial

# 使用方式1: 直接装饰
@tool
def search_web(query: str) -> str:
    \"\"\"在互联网上搜索信息

    Args:
        query: 搜索关键词
    \"\"\"
    # 实现搜索逻辑
    return f"搜索结果: {query}"

# 使用方式2: 指定名称
@tool("web_search")
def my_search_function(query: str) -> str:
    \"\"\"搜索工具\"\"\"
    return search_api(query)

# 使用方式3: 带参数
@tool(return_direct=True, parse_docstring=True)
def calculator(expression: str) -> float:
    \"\"\"计算数学表达式

    Args:
        expression: 数学表达式,如"2+2*3"
    \"\"\"
    return eval(expression)

# 使用方式4: 结构化参数
@tool
def send_email(to: str, subject: str, body: str) -> str:
    \"\"\"发送电子邮件

    Args:
        to: 收件人邮箱地址
        subject: 邮件主题
        body: 邮件正文
    \"\"\"
    # 发送逻辑
    return "邮件已发送"

# 使用方式5: 异步工具
@tool
async def async_search(query: str) -> str:
    \"\"\"异步搜索工具\"\"\"
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/search?q={query}")
        return response.text

# 在Agent中使用
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
model_with_tools = model.bind_tools([search_web, calculator, send_email])

response = model_with_tools.invoke("帮我搜索Python教程")
# AIMessage(tool_calls=[{"name": "search_web", "args": {"query": "Python教程"}, "id": "..."}])

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Decorator as @tool装饰器
    participant Factory as _tool_factory
    participant Struct as StructuredTool
    participant Model as ChatModel

    User->>Decorator: @tool<br/>def search(query: str): ...
    Decorator->>Decorator: 提取函数名"search"
    Decorator->>Factory: _create_tool_factory("search")
    Factory->>Factory: 从docstring提取描述
    Factory->>Factory: 检测函数签名
    Factory->>Struct: from_function(...)<br/>创建StructuredTool
    Struct->>Struct: 生成args_schema<br/>从类型注解
    Struct-->>Decorator: StructuredTool实例
    Decorator-->>User: search工具实例

    User->>Model: bind_tools([search])
    Model->>Model: 转换为模型特定格式
    Model-->>User: 绑定工具的模型

    User->>Model: invoke("搜索Python")
    Model-->>User: AIMessage(tool_calls=[...])

input: str | dict | ToolCall  # 工具输入
config: RunnableConfig | None  # 运行配置

Any  # 工具返回值,类型取决于工具实现

class BaseTool(RunnableSerializable):
    def invoke(
        self,
        input: str | dict | ToolCall,
        config: RunnableConfig | None = None,
    ) -> Any:
        """执行工具

        Args:
            input: 工具输入
            config: 运行配置

        Returns:
            工具执行结果
        """
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)

        # 1) 解析输入
        tool_input, tool_call_id = self._parse_input(input)

        # 2) 开始工具调用回调
        run_manager = callback_manager.on_tool_start(
            {"name": self.name, "description": self.description},
            tool_input if isinstance(tool_input, str) else str(tool_input),
        )

        # 3) 执行工具
        try:
            observation = self._run(
                **tool_input if isinstance(tool_input, dict) else {"input": tool_input},
                run_manager=run_manager,
            )
        except ToolException as e:
            # 4) 处理工具异常
            if self.handle_tool_error:
                observation = self._handle_error(e)
            else:
                run_manager.on_tool_error(e)
                raise
        except Exception as e:
            # 5) 处理其他异常
            run_manager.on_tool_error(e)
            raise
        else:
            # 6) 成功回调
            run_manager.on_tool_end(observation)

        return observation

    def _parse_input(self, input: str | dict | ToolCall) -> tuple[Any, str | None]:
        """解析输入"""
        if isinstance(input, str):
            return input, None
        elif isinstance(input, dict):
            if "type" in input and input["type"] == "tool_call":
                # ToolCall字典
                return input["args"], input.get("id")
            else:
                # 普通参数字典
                return input, None
        else:
            # ToolCall对象
            return input.args, input.id

    @abstractmethod
    def _run(self, *args, run_manager: CallbackManagerForToolRun | None = None, **kwargs) -> Any:
        """子类实现的核心逻辑"""
        ...

    def _handle_error(self, error: ToolException) -> str:
        """处理工具错误"""
        if self.handle_tool_error is True:
            return f"Error: {str(error)}"
        elif isinstance(self.handle_tool_error, str):
            return self.handle_tool_error
        elif callable(self.handle_tool_error):
            return self.handle_tool_error(error)

# 直接调用工具
@tool
def calculator(expression: str) -> float:
    \"\"\"计算表达式\"\"\"
    return eval(expression)

result = calculator.invoke("2+2*3")
# 8.0

# 传入字典
@tool
def send_email(to: str, subject: str) -> str:
    \"\"\"发送邮件\"\"\"
    return f"邮件已发送到{to}"

result = send_email.invoke({"to": "user@example.com", "subject": "Hello"})
# "邮件已发送到user@example.com"

# Agent执行工具调用
tool_call = ToolCall(
    name="calculator",
    args={"expression": "2+2"},
    id="call_123"
)

result = calculator.invoke(tool_call)
# 4.0

# 错误处理
@tool(handle_tool_error=True)
def risky_tool(input: str) -> str:
    \"\"\"可能失败的工具\"\"\"
    if input == "bad":
        raise ToolException("Invalid input")
    return f"Success: {input}"

result = risky_tool.invoke("bad")
# "Error: Invalid input" (不抛出异常)

result = risky_tool.invoke("good")
# "Success: good"

sequenceDiagram
    autonumber
    participant Agent
    participant Tool as BaseTool
    participant CB as CallbackManager
    participant Impl as _run实现

    Agent->>Tool: invoke({"query": "Python"})
    Tool->>Tool: _parse_input
    Tool->>CB: on_tool_start

    Tool->>Impl: _run(query="Python")

    alt 执行成功
        Impl-->>Tool: "搜索结果..."
        Tool->>CB: on_tool_end
        Tool-->>Agent: "搜索结果..."
    else 工具异常
        Impl-->>Tool: ToolException
        Tool->>Tool: _handle_error
        Tool->>CB: on_tool_error
        Tool-->>Agent: "Error: ..."
    else 其他异常
        Impl-->>Tool: Exception
        Tool->>CB: on_tool_error
        Tool-->>Agent: 抛出异常
    end

func: Callable | None  # 同步函数
coroutine: Callable | None  # 异步函数
name: str  # 工具名称
description: str  # 工具描述
return_direct: bool  # 是否直接返回
args_schema: ArgsSchema | None  # 参数schema
infer_schema: bool  # 是否推导schema

StructuredTool  # 结构化工具实例

class StructuredTool(BaseTool):
    @classmethod
    def from_function(
        cls,
        func: Callable | None,
        coroutine: Callable | None,
        name: str,
        description: str,
        return_direct: bool = False,
        args_schema: ArgsSchema | None = None,
        infer_schema: bool = True,
        **kwargs,
    ) -> StructuredTool:
        """从函数创建结构化工具

        Args:
            func: 同步函数
            coroutine: 异步函数
            name: 工具名称
            description: 工具描述
            return_direct: 是否直接返回
            args_schema: 自定义参数schema
            infer_schema: 是否推导schema

        Returns:
            StructuredTool实例
        """
        # 1) 确定使用哪个函数
        fn = func or coroutine

        # 2) 生成或使用args_schema
        if args_schema is None and infer_schema:
            args_schema = create_schema_from_function(
                name,
                fn,
                parse_docstring=kwargs.get("parse_docstring", False),
            )

        # 3) 创建实例
        return cls(
            name=name,
            description=description,
            func=func,
            coroutine=coroutine,
            args_schema=args_schema,
            return_direct=return_direct,
            **kwargs,
        )

def create_schema_from_function(
    name: str,
    func: Callable,
    *,
    parse_docstring: bool = False,
) -> type[BaseModel]:
    """从函数签名创建Pydantic schema

    Args:
        name: schema名称
        func: 函数
        parse_docstring: 是否解析docstring获取参数描述

    Returns:
        Pydantic模型类
    """
    # 1) 获取函数签名
    sig = inspect.signature(func)

    # 2) 构建字段定义
    fields = {}
    for param_name, param in sig.parameters.items():
        if param_name in FILTERED_ARGS:
            # 跳过特殊参数(如run_manager)
            continue

        # 提取类型注解
        annotation = param.annotation if param.annotation != inspect.Parameter.empty else Any

        # 提取默认值
        default = param.default if param.default != inspect.Parameter.empty else ...

        # 从docstring提取描述
        description = None
        if parse_docstring:
            description = _extract_param_description(func, param_name)

        fields[param_name] = (annotation, Field(default, description=description))

    # 3) 创建Pydantic模型
    return create_model(f"{name}Schema", **fields)