第 5 章 · 02 生态集成四件套与 ToolCollection

第 5 章 · 02 生态集成四件套与 ToolCollection

本节摘要:上一节讲「怎么造工具」,本节讲「怎么白嫖工具」。ToolCollection 批量管理工具集;生态集成四件套打通四条管道:from_hub 从 HF Hub 拉取别人分享的工具,Tool.push_to_hub 反向分享自己的工具;from_space 把 Gradio Space 直接当工具调用——HF 生态数千个 Space(图像生成、换脸、OCR……)一夜之间全变成 agent 的武器库;from_mcp 经 MCPAdapt 接入 MCP server 暴露的工具(mcp_client.py 的 MCPClient 负责连接管理);from_langchain 与 LangChain 工具互转。战略只有一个:一次编写,处处复用,蹭满 HF 生态。

内容来源:src/smolagents/tools.py、src/smolagents/mcp_client.py

⚠️ 注意:from_hub 与 from_mcp 的文档都带同一条警告——加载工具意味着「下载代码并在本地执行」,务必像审查 pip 包一样审查工具代码,并显式传 trust_remote_code=True 确认你信任来源。

学习目标

1. 掌握 ToolCollection 的两种来源(from_hub/from_mcp)。
2. 读懂 from_space 的运行机制(gradio_client 调用 + API 自动发现)。
3. 理解 MCPClient 的上下文管理器生命周期。
4. 会用 from_langchain 做双向兼容。

一、ToolCollection:工具的集装箱

ToolCollection 本体极简——就是一层包装:

class ToolCollection:
    def __init__(self, tools: list[Tool]):
        self.tools = tools
​

价值在两个类方法工厂。from_hub 从 HF Hub 的 collection(合集)一次拉一整套:

@classmethod
def from_hub(cls, collection_slug: str, token: str | None = None,
             trust_remote_code: bool = False) -> "ToolCollection":
    _collection = get_collection(collection_slug, token=token)
    _hub_repo_ids = {item.item_id for item in _collection.items if item.item_type == "space"}
    tools = [Tool.from_hub(repo_id, token, trust_remote_code) for repo_id in _hub_repo_ids]
    return cls(tools)
​

官方示例一条链路拉起图像生成工具集:

image_tool_collection = ToolCollection.from_hub(
    "huggingface-tools/diffusion-tools-6630bb19a942c2306a2cdb6f"
)
agent = CodeAgent(tools=[*image_tool_collection.tools], add_base_tools=True)
agent.run("Please draw me a picture of rivers and lakes.")
​

注意细节:collection 里只挑 item_type == "space" 的条目——往合集里放模型和数据集纯属展示,不会误拉。单个工具的 Tool.from_hub 流程是:强制 trust_remote_code 确认 → hf_hub_download 下载 tool.py → Tool.from_code 在动态模块里 exec 源码、找到 Tool 子类、实例化。反向分享用 Tool.push_to_hub(repo_id):把工具源码(tool.py)、自动生成的 Gradio 演示(app.py)和依赖清单(requirements.txt)打包成一个 Space 上传——上一节 tool_validation 保证的「自包含」就是为这一步服务的。

二、from_space:数千个 Gradio Space 变工具

from_space 是四件套里最「生态杠杆」的一个——HF Hub 上有数千个公开 Space,每个都是别人部署好的交互式应用,from_space 让 agent 直接调用它们:

>>> image_generator = Tool.from_space(
...     space_id="black-forest-labs/FLUX.1-schnell",
...     name="image-generator",
...     description="Generate an image from a prompt"
... )
>>> image = image_generator("Generate an image of a cool surfer in Tahiti")
​

无需本地 GPU、无需下载模型权重——FLUX 图像生成直接当函数调。实现靠 gradio_client,__init__ 里自动发现 Space 的 API 并据此生成工具的 inputs schema:

def __init__(self, space_id, name, description="", api_name=None, token=None):
    self.name = name
    self.description = description
    self.client = Client(space_id, hf_token=token)
    space_api = self.client.view_api(return_format="dict", print_info=False)
    space_description = space_api["named_endpoints"]
    if api_name is None:                    # 未指定时取第一个可用端点
        api_name = list(space_description.keys())[0]
    ...
    self.inputs = {}
    for parameter in space_description_api["parameters"]:
        parameter_type = parameter["type"]["type"]
        if parameter_type == "object":
            parameter_type = "any"
        self.inputs[parameter["parameter_name"]] = {
            "type": parameter_type,
            "description": parameter["python_type"]["description"],
            "nullable": parameter["parameter_has_default"],
        }
    output_component = space_description_api["returns"][0]["component"]
    self.output_type = "image" if output_component == "Image" else (
        "audio" if output_component == "Audio" else "any")
​

view_api 拿到 Space 暴露的端点、参数名、类型、描述,自动拼出 inputs;输出组件是 Image 就声明 output_type 为 image。调用时 forward 把参数喂给 client.predict:

def forward(self, *args, **kwargs):
    args = list(args)
    for i, arg in enumerate(args):
        args[i] = self.sanitize_argument_for_prediction(arg)
    ...
    output = self.client.predict(*args, api_name=self.api_name, **kwargs)
    ...
    if isinstance(output, str) and any([output.endswith(ext) for ext in IMAGE_EXTENSIONS]):
        output = AgentImage(output)
    elif isinstance(output, str) and any([output.endswith(ext) for ext in AUDIO_EXTENSIONS]):
        output = AgentAudio(output)
    return output
​

sanitize_argument_for_prediction 处理参数的地毯式转换:PIL 图片先存成临时文件、本地路径与 URL 都转成 gradio_client 的 handle_file 句柄;返回值按扩展名(.png/.jpg…/.mp3/.wav…)包装成 AgentImage/AgentAudio——多模态输出无缝回流到 agent 的记忆里(还记得 ActionStep.observations_images 吗)。

三、from_mcp:MCP server 工具接入

MCP(Model Context Protocol)是工具生态的通用协议,smolagents 经 mcpadapt 适配接入。两条入口:ToolCollection.from_mcp(上下文管理器)与 MCPClient(手动管理连接)。先看官方 Stdio 示例:

server_parameters = StdioServerParameters(
    command="uvx",
    args=["--quiet", "pubmedmcp@0.1.3"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

with ToolCollection.from_mcp(server_parameters, trust_remote_code=True) as tool_collection:
    agent = CodeAgent(tools=[*tool_collection.tools], add_base_tools=True, model=model)
    agent.run("Please find a remedy for hangover.")
​

from_mcp 的实现:

@classmethod
@contextmanager
def from_mcp(cls, server_parameters, trust_remote_code=False, structured_output=None):
    ...
    if isinstance(server_parameters, dict):
        transport = server_parameters.get("transport")
        if transport is None:
            transport = "streamable-http"
        if transport not in {"sse", "streamable-http"}:
            raise ValueError(f"Unsupported transport: {transport}. ...")
    if not trust_remote_code:
        raise ValueError(
            "Loading tools from MCP requires you to acknowledge you trust the MCP server, "
            "as it will execute code on your local machine: pass `trust_remote_code=True`."
        )
    with MCPAdapt(server_parameters, SmolAgentsAdapter(structured_output=structured_output)) as tools:
        yield cls(tools)
​

支持两种传输:Stdio(子进程标准输入输出,传 StdioServerParameters)与 HTTP({"url": ..., "transport": "streamable-http"},旧版 sse 已弃用)。真正干活的是 MCPAdapt + SmolAgentsAdapter——它把 MCP server 声明的每个 tool 自动适配成 smolagents 的 Tool 实例;structured_output=True 可启用 MCP 的 outputSchema 结构化输出支持。

mcp_client.py(171 行)的 MCPClient 是同一能力的面向对象封装,负责连接生命周期:

class MCPClient:
    def __init__(self, server_parameters, adapter_kwargs=None, structured_output=None):
        ...
        self._adapter = MCPAdapt(
            server_parameters, SmolAgentsAdapter(structured_output=structured_output), **adapter_kwargs)
        self._tools: list[Tool] | None = None
        self.connect()                  # 初始化即连接

    def connect(self):
        """Connect to the MCP server and initialize the tools."""
        self._tools: list[Tool] = self._adapter.__enter__()

    def disconnect(self, ...):
        """Disconnect from the MCP server"""
        self._adapter.__exit__(exc_type, exc_value, exc_traceback)

    def get_tools(self) -> list[Tool]:
        if self._tools is None:
            raise ValueError(
                "Couldn't retrieve tools from MCP server, run `mcp_client.connect()` first ..."
            )
        return self._tools
​

初始化即连接(文档强调:不用上下文管理器就务必 try…finally 保证 disconnect),get_tools() 前若未连接会收到明确报错。工具集在会话创建时固定,未来版本才支持动态感知 server 新增工具。

四、from_langchain 与生态战略

LangChain 存量工具也接得进来:

@staticmethod
def from_langchain(langchain_tool):
    class LangChainToolWrapper(Tool):
        skip_forward_signature_validation = True

        def __init__(self, _langchain_tool):
            self.name = _langchain_tool.name.lower()
            self.description = _langchain_tool.description
            self.inputs = _langchain_tool.args.copy()
            for input_content in self.inputs.values():
                if "title" in input_content:
                    input_content.pop("title")     # 剥掉 UI 用的 title 字段
                input_content["description"] = ""
            self.output_type = "string"
            ...
        def forward(self, *args, **kwargs):
            tool_input = kwargs.copy()
            for index, argument in enumerate(args):  # 位置参数归位到第一个输入键
                if index < len(self.inputs):
                    input_key = next(iter(self.inputs))
                    tool_input[input_key] = argument
            return self.langchain_tool.run(tool_input)
​

LangChain 工具自带 args schema(JSON Schema 格式),直接搬过来、剥掉 UI 用的 title 字段即可;调用时把位置参数归位到第一个输入键,再走 langchain_tool.run。注意 skip_forward_signature_validation = True——wrapper 的 forward 是泛化签名,绕过上一节讲的「forward 签名必须与 inputs 严格一致」校验(SpaceToolWrapper 同理)。反向提醒:from_space/from_langchain 包装的工具不可再 save/push——源码无法自包含重建,to_dict 里会直接抛 ValueError。

四件套合起来看,smolagents 的工具生态战略非常清晰:

HF Hub 工具 Space ──from_hub──┐
Gradio Space    ──from_space──┤        ┌── CodeAgent 的 tools 参数
MCP server      ──from_mcp────┼──────→ │
LangChain 工具  ──from_langchain┘      └── ToolCollection 批量注入
自己写的 Tool/@tool ──push_to_hub 反哺生态
​

你不需要为 agent 重写任何已有能力:Hub 上有现成工具就拉,Space 能干活就包,MCP server 有协议就接。写好一个工具 push 上去,全网 smolagents 用户都能复用——工具的网络效应远大于单体框架。

💡 阶梯要点:本阶拿到「生态」装备。四个方向各有其位:from_hub 拉代码级工具(本地执行,要 trust_remote_code),from_space 拉服务级工具(远程执行,零 GPU 成本),from_mcp 走通用协议,from_langchain 兼容存量生态。共同出口都是 Tool 对象——上一节的基类抽象保证了四条管道殊途同归。

本节要点回顾

• ToolCollection:tools 列表的薄包装;from_hub 按 collection_slug 批量拉取(只取 space 条目)。
• Tool.from_hub:下载 tool.py → exec 重建 → 实例化;必须显式 trust_remote_code=True;push_to_hub 上传 tool.py+app.py+requirements.txt。
• from_space:gradio_client 的 view_api 自动发现参数生成 inputs schema,forward 走 predict,文件/URL 自动转换,输出按扩展名包成 AgentImage/AgentAudio。
• from_mcp:MCPAdapt+SmolAgentsAdapter 适配,支持 Stdio 与 streamable-http 传输,上下文管理器管理生命周期;MCPClient 是 OO 封装,init 即 connect。
• from_langchain:直接复用其 args schema,泛化签名跳过 forward 校验;wrapper 类工具不可二次序列化。

下一节:03 内置工具与 GradioUI——开箱即用的 DuckDuckGo 搜索、网页访问、维基百科查询,以及一行 launch() 起服务的交互界面。