LangChain工具开发：自定义Tools与Schema定义（附：Python沙箱安全执行）

作者： Dr.n8n 更新时间：2025-12-19 20:00:41 分类：n8n教程

当LangChain的现成工具不够用时，你该怎么办？

上周帮一家跨境电商客户搭建智能客服Agent，他们需要实时调用内部ERP系统查询库存——但LangChain官方工具库里根本没有这个接口。更糟的是，他们还想让运营人员能“写点Python脚本”自定义逻辑，又怕把服务器搞崩。这其实是两个高频痛点：一是工具不够灵活，二是执行环境不安全。今天我就手把手教你解决。

自定义Tools的本质：给AI配个“瑞士军刀”

LangChain里的Tool，说白了就是一段带描述的函数。AI（比如ChatGPT）看到的是“功能说明书”，执行时调用的是你的代码。就像你雇了个助理，你得明确告诉他：“这是开瓶器（功能名），专门用来开红酒瓶盖（描述），操作方法是逆时针旋转（函数逻辑）。”

我在项目里踩过坑：千万别只写函数不写Schema！否则AI根本不知道参数该怎么传。下面是最小可行示例：

from langchain.tools import tool

@tool
def query_inventory(product_id: str) -> str:
    """根据商品ID查询实时库存，返回JSON格式字符串。"""
    # 模拟调用ERP接口
    return f'{"product_id": "{product_id}", "stock": 150}'

Schema定义不是玄学，而是“API说明书”

很多教程只讲装饰器@tool，却不说背后的Schema机制。其实LangChain会自动从函数签名和Docstring生成OpenAPI风格的Schema。但自动推断常出错——比如把int识别成str，或漏掉必填字段。这时候就得手动干预。

实战经验：我曾因没指定参数类型，导致AI传了"null"字符串而非None，引发下游系统崩溃。现在所有自定义Tool我都会显式声明Schema。

from langchain_core.pydantic_v1 import BaseModel, Field

class InventoryInput(BaseModel):
    product_id: str = Field(description="商品唯一标识符，如SKU-2024")
    warehouse_code: str = Field(default="WH01", description="仓库编码，默认主仓")

@tool(args_schema=InventoryInput)
def query_inventory(product_id: str, warehouse_code: str = "WH01") -> str:
    # ... 实际业务逻辑

Python沙箱：让非技术人员也能安全写脚本

客户总想“自己加点逻辑”，直接exec()用户输入的Python代码？等于在服务器上贴“欢迎黑客”的标语。我的解决方案是三层防护：

语法树过滤：用ast模块解析代码，禁止import、open()等危险操作
超时熔断：任何脚本超过3秒强制终止
资源隔离：在Docker容器内运行，内存限制256MB

下面是个简化版沙箱实现：

import ast
import signal
from types import FrameType

class RestrictedExec:
    def __init__(self, timeout=3):
        self.timeout = timeout
    
    def execute(self, code_str: str, context: dict) -> any:
        # 第一步：AST检查
        tree = ast.parse(code_str)
        for node in ast.walk(tree):
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                raise ValueError("禁止导入模块")
        
        # 第二步：设置超时
        def timeout_handler(signum: int, frame: FrameType):
            raise TimeoutError("脚本执行超时")
        
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(self.timeout)
        
        try:
            # 在受限上下文中执行
            exec_globals = {"__builtins__": {}}
            exec_locals = {}
            exec(code_str, exec_globals, exec_locals)
            return exec_locals.get('result')
        finally:
            signal.alarm(0)