个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain 自定义 Tool 的三种写法装饰器、StructuredTool 与继承 BaseTool一、三种写法速览二、写法一tool 装饰器2.1 参数校验用 Pydantic三、写法二StructuredTool.from_function3.1 参数 schema 同样可以指定四、写法三继承 BaseTool4.1 安全是硬要求五、三种写法对比六、工具组织的一些经验6.1 数量控制在 10 个以内6.2 名字要有区分度6.3 返回值要模型友好6.4 用 ToolNode 接入图七、小结LangChain 自定义 Tool 的三种写法装饰器、StructuredTool 与继承 BaseTool上一篇讲了 Tool Calling 的完整数据流工具是用tool装饰器定义的。但真实项目里工具没那么简单有的要异步、有的要访问数据库、有的参数需要严格校验、有的还得带上运行时配置。这篇把三种写法和它们的适用边界一次讲清。一、三种写法速览写法代码量适用场景tool装饰器最少80% 的场景同步函数直接转工具StructuredTool.from_function中等需要async、需要分开指定同步/异步实现继承BaseTool最多需要复用连接、持有状态、精细控制错误处理默认用装饰器需要 async 用 StructuredTool要持有资源数据库连接、客户端才继承 BaseTool。二、写法一tool 装饰器fromlangchain_core.toolsimporttooltooldefcalculate(expression:str)-str:计算数学表达式。只支持加减乘除和括号不支持函数调用。try:returnstr(eval(expression,{__builtins__:{}},{}))exceptExceptionase:returnf计算失败{e}如果默认推导的名字/描述不满意可以显式指定tool(calculator,description计算数学表达式支持 - * / 和括号,return_directFalse)defcalculate(expression:str)-str:...几个参数第一个位置参数覆盖工具名默认用函数名description覆盖描述默认用 docstringreturn_directTrue工具结果直接作为最终答案返回不再经过模型。适合发送邮件下单这类执行型工具避免模型再画蛇添足复述一遍。2.1 参数校验用 Pydantic类型注解只能表达基础类型。要做范围、枚举、格式校验用args_schemafrompydanticimportBaseModel,FieldfromtypingimportLiteralclassSearchInput(BaseModel):搜索参数query:strField(description搜索关键词2~50 字,min_length2,max_length50)top_k:intField(description返回结果条数,ge1,le20,default5)order:Literal[relevance,time]Field(description排序方式relevance按相关性time按时间,defaultrelevance)tool(args_schemaSearchInput)defsearch(query:str,top_k:int5,order:strrelevance)-str:在企业知识库中搜索文档。returnf搜索 {query}返回{top_k}条按{order}排序Pydantic 里的约束会一并写进送给模型的 schema模型看到ge1, le20和枚举值就不太会传离谱的参数了。三、写法二StructuredTool.from_function装饰器的短板是不方便定义异步实现。StructuredTool.from_function可以同时给同步和异步两个版本importasyncioimportaiohttpfromlangchain_core.toolsimportStructuredTooldeffetch_sync(url:str)-str:同步抓取网页importrequestsreturnrequests.get(url,timeout10).text[:500]asyncdeffetch_async(url:str)-str:异步抓取网页asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url,timeout10)asresp:return(awaitresp.text())[:500]fetch_toolStructuredTool.from_function(funcfetch_sync,# 同步实现coroutinefetch_async,# 异步实现namefetch_webpage,description抓取指定 URL 的网页正文返回前 500 字。,)有了coroutine就可以resultawaitfetch_tool.ainvoke({url:https://example.com})在 Agent 里这一点很重要——如果 Agent 一次要调 3 个工具有异步实现就能并发执行总耗时从 3 秒降到 1 秒。如果只给了funcainvoke会退化成在线程池里跑同步函数能并发但占线程两者都不给就报错。3.1 参数 schema 同样可以指定frompydanticimportBaseModel,FieldclassFetchInput(BaseModel):url:strField(description要抓取的网页地址必须是 http/https 开头)fetch_toolStructuredTool.from_function(funcfetch_sync,coroutinefetch_async,namefetch_webpage,description抓取指定 URL 的网页正文。,args_schemaFetchInput,)四、写法三继承 BaseTool什么时候必须继承当工具需要持有资源或状态。典型场景数据库连接池、Redis 客户端、HTTP 长连接池、带认证的第三方 SDK——这些都不该每次调用都新建。fromlangchain_core.toolsimportBaseToolfrompydanticimportBaseModel,FieldfromtypingimportOptional,Typeimportsqlite3classSQLQueryInput(BaseModel):sql:strField(description要执行的 SELECT 语句)classSQLQueryTool(BaseTool):name:strquery_databasedescription:str(查询业务数据库。只支持 SELECT 语句禁止 INSERT/UPDATE/DELETE。\n可用表orders(id, user_id, amount, status, created_at)\n users(id, name, city, vip_level))args_schema:Type[BaseModel]SQLQueryInput return_direct:boolFalsedb_path:str# 自定义字段Pydantic 会管理_conn:Optional[sqlite3.Connection]None# 私有字段下划线开头def_run(self,sql:str)-str:同步执行ifnotsql.strip().lower().startswith(select):return只允许 SELECT 语句try:connself._get_conn()rowsconn.execute(sql).fetchall()returnstr(rows[:20])# 截断别把整个表吐给模型exceptExceptionase:returnf查询失败{e}asyncdef_arun(self,sql:str)-str:异步执行这里简单地在线程池里跑同步版本importasyncioreturnawaitasyncio.to_thread(self._run,sql)def_get_conn(self):ifself._connisNone:self._connsqlite3.connect(self.db_path)returnself._conn sql_toolSQLQueryTool(db_pathdata/app.db)几个要点必须实现_run可选实现_arun私有属性用下划线开头_conn否则 Pydantic 会尝试校验它的类型name/description/args_schema写成类属性并标注类型业务配置如db_path写成普通字段实例化时传入。4.1 安全是硬要求数据库类工具必须做这几件事缺一不可def_run(self,sql:str)-str:ssql.strip().lower()# 1. 只允许 SELECTifnots.startswith(select):return拒绝执行只允许 SELECT# 2. 封掉危险关键词forkwin[drop,delete,update,insert,alter,attach,pragma]:ifkwins:returnf拒绝执行包含危险关键词{kw}# 3. 强制加 LIMIT避免全表扫描iflimitnotins:sqlsql.rstrip(;) LIMIT 100# 4. 超时保护...让模型能执行 SQL 是危险的上面的检查能挡住大部分误操作但生产环境更稳妥的做法是不开放任意 SQL只开放若干个参数化的查询工具如get_order_by_id、count_orders_by_status把灵活换成可控。五、三种写法对比维度toolStructuredToolBaseTool代码量3 行10 行40 行异步实现❌✅coroutine✅_arun持有连接/状态❌❌✅Pydantic 校验✅args_schema✅✅复用/继承❌❌✅六、工具组织的一些经验6.1 数量控制在 10 个以内工具越多模型越容易选错。超过 10 个工具选择准确率会明显下降。如果业务确实需要几十个能力考虑按场景分组动态注入用户进入订单场景才给订单相关工具用一个路由工具先分类再加载对应工具集。6.2 名字要有区分度# ❌ 模型分不清search_doc/search_web/search_user# ✅ 用途一目了然search_internal_wiki/search_internet/lookup_user_profile6.3 返回值要模型友好工具返回什么模型就看到什么。几条建议返回文本不要返回对象模型读不懂 Python 对象截断长结果别把 1000 行数据全塞进去带上结构提示比如共 3 条1) ... 2) ... 3) ...出错时返回可读的错误描述模型可能据此修正参数重试。6.4 用 ToolNode 接入图定义好的工具列表在 LangGraph 里直接用ToolNodefromlanggraph.prebuiltimportToolNode tools[sql_tool,fetch_tool,search_tool]tool_nodeToolNode(tools)它会自动处理取 tool_calls → 执行 → 包成 ToolMessage这一整套循环。七、小结默认tool需要 async 用StructuredTool.from_function给coroutine要持有连接/状态才继承BaseTool参数校验一律用 Pydanticargs_schema约束会写进 schema 帮模型少犯错继承BaseTool时私有属性必须下划线开头否则 Pydantic 会报错数据库工具必须做SELECT 白名单 危险词拦截 LIMIT 超时更稳妥的是只开放参数化的固定查询工具数量控制在 10 个以内多了模型会选错返回值要截断、要文本化、出错要返回可读信息。下一篇讲结构化输出——让模型稳定吐出符合 schema 的 JSON。