嘿伙计！

今天聊聊怎么把 LangChain 工具打包成你自己的 AI 助手。

说实话，一开始我也觉得这个名字听起来像玩具，但实际上它Neng让你把大模型变成真正的工作帮手。

咱们先从Zui基础的“为什么要自定义工具”说起。

LangChain 的默认工具挺多的。

比如搜索引擎、计算器之类。

但业务场景往往geng复杂。

比如你想让模型直接读取数据库里的表格。

或者把本地文档内容送进去生成摘要。

这时就得自己写一个专属工具了。

① Neng让 LLM 真正操作外部资源。

② 减少调用次数，提高效率。

③ 易于复用和维护。

windows 或 macOS dou行，只需 pip 安装即可。哈哈！

pip install langchain langchain-openai langchain-community 
# Ru果你想用向量数据库，还需要安装 FAISS 或 Milvus 等。

Baidu 的算法对技术深度要求高，且偏好中文原创内容；而我们的文章使用了大量代码示例和英文库名，这种混合风格可Neng导致被算法判为“技术过硬但缺少生活化语境”，从而降权。再加上Zui近 Baidu 搜索指数下降，让人懊恼。但别担心，你Ke以把它发布到技术社区或博客平台，让geng多人kan到。害！

from abc import ABCMeta, abstractmethod
class BaseTool:
    """一个通用工具基类，所有自定义 Tool 必须继承它。"""
    def __init__:
        self.name=name          # 工具唯一标识符，如 get_user_info  
        self.description=dsc     # 简短描述，用于提示 LLM  
        self.schema=schema      # 参数校验结构  
    @abstractmethod  
    def execute:  
        """核心业务逻辑，返回字符串或 JSON 字符串。"""  
    def before_execute:  
        print  
    def after_execute:  
        print}")  
    def on_error:  
        return f"{self.name} 执行失败：{str}"
    def build_tool:  
        """包装为 LangChain 可调用的 Tool 对象。””
        # 内部实现细节隐藏在这里直接返回可用对象即可。
"""

class CalculatorTool:
    """简单数学计算器。"""
    def __init__:  
        super.__init__(
            name="calculator",
            dsc="接受表达式并返回结果",
            schema={
                "expression": str,
                # 用 dict 定义参数类型与描述，仅演示，不Zuo严格校验。
            }
        )
    def execute:  
        expr = args.get
        try:
            result = eval
            return str
        except Exception as exc:
            raise RuntimeError
# 注册后可直接使用：
calc_tool = CalculatorTool.build_tool

N.B.: 在生产环境里绝不Neng直接 eval，请使用安全解析器或专门库。哈哈！我自己dou没注意到这一点，现在改了下。 不对不对，该改为安全方案哦！

class CodeFormatter:
    """简易 JS/TS/JSON 格式化器。”
    它会把所有行前面补齐指定空格数。
"""
    def __init__:  
        super.__init__(
            name="formatter",
            dsc="给前端代码Zuo排版",
            schema={
                "source": str,
                "indent": int,
            }
        )
    @staticmethod    
    def _simple_indent:  
        indent = " "*line_no 
...
"""
****
说明：
NoSQL 数据库里也经常要把 JSON 转 CSV；这个工具Ke以双向转换，但为了演示，我只写了核心逻辑，不再贴完整代码。不过你Ke以复制粘贴进项目里跑一下。懂吗？嗯哼~ 
六、多场景实例集合

- 前端开发助手：formatter、file_reader 等。
- 数据处理：converter 、file_reader 等。
- 网络抓取：web_fetcher 。 
- 系统运维：日志分析器。 
- 文档提炼：doc_summarizer。 
- …还有geng多，你Ke以随意
！ 

案例一——文件读取 Tool

import os
class FileReader:
    MAX_SIZE = 1024*1024   # Zui大 1MB
    def __init__:   
         super.__init__(
             name='file_reader',
             dsc='读取项目目录内文件',
             schema={'path':str}
         )
     def execute:   
         path=args.get
         abs_path=os.path.abspath
         root=os.getcwd
         if not abs_path.startswith:
             raise ValueError
         size=os.path.getsize
         if size>FileReader.MAX_SIZE:
             raise ValueError
         data=open.read
         # 内容太长截断处理：
         max_len=50000 
         if len>max_len:
             data=data+"\
\
..."
         return data
# 注册：
file_reader_tool=FileReader.build_tool

案例二——网页抓取 Tool

class WebFetcher:
     ALLOWED_DOMAINS=
     def __init__:
          super.__init__(
              name='web_fetcher',
              dsc='抓取网页文本',
              schema={'url':str,'extract':bool}
          )
     asyncdef fetch_content->str:
          await asyncio.sleep           # 模拟网络延迟       
          return f"{url} 内容模拟数据..."
     @staticmethod    
     asyncdef _clean_text->str:
          import re     
          txt=re.sub
          return re.sub.strip
     asyncdef execute:
          url=args.get
          extract=args.get
          try:
              new_url=URL
          except Exception:
              raise ValueError
          host=new_url.split.split
          if host not in WebFetcher.ALLOWED_DOMAINS:
              raise ValueError
          raw_html=await WebFetcher.fetch_content
          text=self._clean_text
          return text   # 截断防止过长
# 注册：
web_fetcher_tool=WebFetcher.build_tool

注

#URL 是 python 内置对象，用来验证地址是否正确；Ru果你没安装 urllib.parse，Ke以自行实现简单检查。
#WebFetcher 的 fetch_content 用的是异步函数，要在主程序里配合 asyncio.run 调用才行哦！ 
#Ru果你想真正抓取页面请替换 fetch_content 为 requests.get 或 aiohttp 请求实际地址。 
#记得在运行前先 pip install aiohttp 。不过这段代码Yi经足够演示思路啦～ 
#别忘了加权限控制，否则别人Neng随便抓站点……嗨哟~ 
#另外Ru果你要支持 HTTPS，需要先安装 certifi 并在请求时验证证书，否则可Neng报 SSL 错误。 嗯哼～ 
      
案例三——CSV 与 JSON 双向转换 Tool

import csv,json 
class Converter:
     def __init__:
           super.__init__(
               name='converter',
               dsc='JSON↔CSV 转换',
               schema={'data':str,'mode':'json_to_csv'}
           )
     @staticmethod    
     asyncdef json_to_csv->str:
           try:
               arr=json.loads
           except json.JSONDecodeError:
               raise ValueError
           if not isinstance or len==0:
               raise ValueError
           fieldnames=list)
           output=
           writer=csv.DictWriter
           writer.writeheader
           writer.writerows
           return "\
".join
     @staticmethod    
     asyncdef csv_to_json->str:
           reader=csv.DictReader)
           rows=
           return json.dumps
     asyncdef execute:
           mode=args.get
           data=args.get
           if mode=='json_to_csv':
                return await Converter.json_to_csv
           elif mode=='csv_to_json':
                return await Converter.csv_to_json
           else:
                raise ValueError
# 注册：
converter_tool=Converter.build_tool

七、异常处理与健壮性设计思路

- 所有外部调用dou包裹 try‑except；捕获后返回友好错误信息，而不是裸抛出异常给用户kan见堆栈帧。.
.
...

**示例代码片段**：

python
asyncdef robust_execution->dict:
      start=time.time
      try:
            res=await tool.execute
      except Exceptionas err :
            res=f"错误:{err}"
      elapsed=time.time-start
      return {"success":bool, "output":res,"time_ms":elapsed}

**关键点回顾**：

- 参数校验一定要Zuo好，否则 LLM 会把错误信息直接吐出来让用户失望。。哈哈～.
.
...



八、动态注册与统一管理机制示例

class ToolRegistry:
   """字典型容器，但提供额外方法方便查询。””
   @classmethod    
   asyncdef register:
         cls=tfunc
   @classmethod    
   asyncdef get:
         return cls.get
   @classmethod    
   asyncdef list_tools:
         return list)

await ToolRegistry.register
await ToolRegistry.register
print)   # 

**说明**：

- 当你的项目越来越多时用 Registry Ke以避免全局变量堆砌，让每个模块只关心自己的业务逻辑。
.
...

九、整合 Demo —— 一个小巧的智Neng助理
下面给出完整 runnable 示例，其中包含：

初始化 LLM；
注入我们刚才写好的四个自定义 Tool；
简单状态机让 LLM 决定何时调用哪个工具；
Zui终将回答呈现给用户。


python
import os,re,json,time,itertools,csv,string,argparse,tempfile 
from dotenv import load_dotenv  
loaddotenv   # 加载环境变量，例如 OPENAIAPI_KEY  
from langchain.chat_models import ChatOpenAI  
from langchain.agents import initialize_agent , AgentType  
from langchain.tools import BaseToolkit  
calctool = CalculatorTool.buildtool
fmttool = CodeFormatter.buildtool
frtfile = FileReader.buildtool
wffetch = WebFetcher.buildtool
tools=
model = ChatOpenAI(
      temperature=0,
      modelname=os.getenv,
      openaiapi_key=os.getenv
)
agentexecutor = initializeagent(
      tools,
      model,
      agenttype=AgentType.ZEROSHOTREACTDESCRIPTION,
      verbose=True,
)
prompt="""
请帮我完成以下任务：
① 使用 calc Zuo一次加法：*56 /78。
② 将下面一段 JavaScript 粘贴进去进行格式化，并保持两空格缩进。
③ 把当前目录下 README.md 的内容读出来并输出前200字符。
④ 抓取 https://siteA/example.html 并提取正文文本。
注意：
每一步dou要分步骤说明，并给出对应的结果。
"""
response = agent_executor.run
print
print


小结与后续方向


性——只需编写新的类继承 BaseTool，然后注册即可；整个系统保持一致接口，无缝切换新旧版本。
安全性——通过白名单域名控制网络抓取，通过路径检查限制文件访问；再加上超时和重试机制，让服务geng稳健。
性Neng优化——Ru果某些操作耗时较长，Ke以拆成异步任务并发执行，也Ke以缓存热点数据降低重复调用成本。


十、常见坑踩点与个人经验分享


  
坑
  
避免方式



  
eval 安全问题
  
用专门解析库或自行限制合法字符


  
递归调用无限循环
  
给每次调用设置Zui大递归深度


  
路径遍历攻击
  
用 os.path.abspath 判断是否位于根目录


  
数据过大导致内存爆炸
  
设置大小阈值并Zuo流式读取


  
没有显式错误信息导致调试困难
  
每个 catch 块dou返回可读字符串


  
未Zuo类型校验导致 LLM 调用无效参数
  
在 Schema 中明确列出必填字段与类型




十一、自测与迭代建议
在正式上线之前，我通常会：
1️⃣ 写几个单元测试覆盖各条边界情况。
✅ 验证流程是否符合预期，并打印详细日志供排查。
🔍 若发现性Neng瓶颈，就针对耗时函数Zuo异步改造或缓存策略。
💡 当用户反馈有误报或缺陷时我会及时修复并回滚版本，以保证体验稳定。

一句话
LangChain 的 Custom Tools 就像是你手中的乐高积木，你只需掌握搭建方法，再根据业务需求拼接不同块子，就Neng轻松打造属于自己的 AI 助手。而且记住一旦习惯这种模式，其它框架迁移也跟着变得容易多了。
所以下次当你面对“我要让模型读表格”或者“我想让 AI 自动抓网页”之类的问题时就去找我的仓库复制粘贴一下吧，说真的，它们Yi经准备好了只等你来玩！
祝编码愉快 🚀

                        

                        

                         

                    

                    

                        

                            标签：  
                            自定义  
                        
                    
                    

                    
                    

                        

                            

                                 上一篇： 
                                CSDN新AI营销，捞钱还是烂货？  
                            
                            

                                 下一篇： 
                                能了解中关村软件园如何为企业服务聚集再创新？