纲要

• 工具在智能体中的核心地位
• 设计原则
  • 最小颗粒度与解耦
  • 结构化参数：Pydantic模型绑定
  • 工具描述与@tool装饰器
• 四大工具实现详解
  • 在线搜索：SerpAPI封装
  • 知识库检索：Chroma向量库 + 查询重写
  • 钉钉待办：情感阈值触发创建
  • 钉钉日历：增删查改功能
• 钉钉 API 客户端封装
• 完整可运行示例
• 总结与相关度说明

工具：智能体的能力边界

如果说大模型是智能体的大脑，那么工具就是它的双手。一个智能体能做什么、做得多好，几乎完全取决于所集成的工具。在 LangChain 中，工具可以是一个函数、一个 API 调用、一个数据库查询——本质上，任何能够接收输入并返回结果的外部能力都可以封装为工具。

构建工具时，有三个核心原则值得反复思考：

1. 最小颗粒度：将功能拆分为原子操作，让智能体自己组合，而不是代劳组合逻辑。
2. 结构化输入：强制工具的入参拥有明确的类型和描述，减少大模型传参随机性导致的错误。
3. 清晰的功能描述：每个工具必须附带精确的注释，这些注释将被注入提示词，成为模型选择工具的唯一依据。

工具集概览

小浪助手作为钉钉智能客服，集成了四类工具：

工具的实现

环境准备

依赖安装：

pipinstalllangchain langchain-openai langchain-community chromadb pydantic python-dotenv google-search-results requests

项目结构：

tools_project/ ├── config.py ├── dingtalk_client.py ├── tools.py ├── main.py └── .env

配置管理config.py

# config.pyimportosfromdotenvimportload_dotenv load_dotenv()classConfig:OPENAI_API_KEY=os.getenv("OPENAI_API_KEY")OPENAI_BASE_URL=os.getenv("OPENAI_BASE_URL","https://api.openai.com/v1")SERPAPI_API_KEY=os.getenv("SERPAPI_API_KEY")DINGTALK_APP_KEY=os.getenv("DINGTALK_APP_KEY")DINGTALK_APP_SECRET=os.getenv("DINGTALK_APP_SECRET")DINGTALK_AGENT_ID=os.getenv("DINGTALK_AGENT_ID")CHROMA_PERSIST_DIR="./chroma_db"

钉钉客户端封装dingtalk_client.py

钉钉 API 大多需要access_token，我们封装一个简单的客户端负责获取和缓存。

# dingtalk_client.pyimportrequestsimporttimefromconfigimportConfigclassDingTalkClient:def__init__(self):self.app_key=Config.DINGTALK_APP_KEY self.app_secret=Config.DINGTALK_APP_SECRET self.agent_id=Config.DINGTALK_AGENT_ID self._token=Noneself._expire_time=0defget_access_token(self)->str:ifself._tokenandtime.time()<self._expire_time:returnself._token url="https://oapi.dingtalk.com/gettoken"params={"appkey":self.app_key,"appsecret":self.app_secret}resp=requests.get(url,params=params)data=resp.json()ifdata.get("errcode")==0:self._token=data["access_token"]self._expire_time=time.time()+7000# 官方有效期7200秒，提前200秒刷新returnself._tokenelse:raiseException(f"获取钉钉token失败:{data}")

工具定义tools.py

这里包含搜索、知识库、待办、日历的全部工具。注意每个函数都用@tool装饰，并附上清晰的描述。

# tools.pyimportosimportjsonfromtypingimportList,Optionalfromdatetimeimportdatetimefromlangchain.toolsimporttoolfromlangchain_community.utilitiesimportSerpAPIWrapperfromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain_community.vectorstoresimportChromafromlangchain.promptsimportChatPromptTemplatefrompydanticimportBaseModel,FieldfromconfigimportConfigfromdingtalk_clientimportDingTalkClient# ---------- Pydantic 参数模型 ----------classTodoInput(BaseModel):subject:str=Field(description="待办事项标题")description:Optional[str]=Field(None,description="详细描述")priority:int=Field(20,description="优先级，数字越小越优先，默认20")classCalendarEventInput(BaseModel):summary:str=Field(description="日程标题")start_time:str=Field(description="开始时间，ISO格式，如2025-01-01T10:00:00+08:00")end_time:str=Field(description="结束时间，ISO格式")location:Optional[str]=Field(None,description="地点")description:Optional[str]=Field(None,description="日程描述")classQueryCalendarInput(BaseModel):start_time:str=Field(description="查询时间段的开始，ISO格式")end_time:str=Field(description="查询时间段的结束，ISO格式")classModifyCalendarInput(BaseModel):event_id:str=Field(description="要修改的日程ID")summary:Optional[str]=Field(None)start_time:Optional[str]=Field(None)end_time:Optional[str]=Field(None)classDeleteCalendarInput(BaseModel):event_id:str=Field(description="要删除的日程ID")# ---------- 搜索工具 ----------@tooldefweb_search(query:str)->str:"""在线搜索最新信息。当需要实时数据或未知事实时使用此工具。"""search=SerpAPIWrapper(serpapi_api_key=Config.SERPAPI_API_KEY)returnsearch.run(query)# ---------- 知识库工具 ----------# 假设向量库已预先创建并填充，这里仅演示检索embeddings=OpenAIEmbeddings(openai_api_key=Config.OPENAI_API_KEY,base_url=Config.OPENAI_BASE_URL)vectorstore=Chroma(persist_directory=Config.CHROMA_PERSIST_DIR,embedding_function=embeddings)defrewrite_query(original:str)->List[str]:"""查询重写：生成多个变体以提高召回率"""llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0.3)prompt=ChatPromptTemplate.from_template("将以下用户问题改写为3个不同角度但语义相同的查询，每个查询单独一行，不要编号。\n""问题: {query}\n改写的查询:")chain=prompt|llm result=chain.invoke({"query":original})queries=[q.strip()forqinresult.content.split("\n")ifq.strip()]return[original]+queries@tooldefsearch_knowledge_base(query:str)->str:"""查询内部知识库，获取LangChain等专业知识。"""queries=rewrite_query(query)all_docs=[]forqinqueries:docs=vectorstore.similarity_search(q,k=2)all_docs.extend(docs)# 去重并取前5个最相关的unique_contents=[]seen=set()fordocinall_docs:ifdoc.page_contentnotinseen:seen.add(doc.page_content)unique_contents.append(doc.page_content)return"\n\n".join(unique_contents[:5])# ---------- 钉钉待办工具 ----------ding_client=DingTalkClient()@tool(args_schema=TodoInput)defcreate_todo(subject:str,description:str="",priority:int=20)->str:"""创建钉钉待办事项。当用户要求记录事务、投诉或强烈负面情绪时使用。"""token=ding_client.get_access_token()url="https://api.dingtalk.com/v1.0/todo/users/me/tasks"headers={"x-acs-dingtalk-access-token":token,"Content-Type":"application/json"}body={"subject":subject,"description":description,"priority":priority,"createdTime":int(datetime.now().timestamp()*1000)}resp=requests.post(url,headers=headers,json=body)data=resp.json()ifresp.status_code==200and"id"indata:returnf"已创建待办「{subject}」，优先级{priority}"else:returnf"创建待办失败:{data}"# ---------- 钉钉日历工具 ----------@tool(args_schema=CalendarEventInput)defcreate_calendar_event(summary:str,start_time:str,end_time:str,location:str="",description:str="")->str:"""在钉钉日历中创建新日程。参数均为ISO 8601格式字符串。"""token=ding_client.get_access_token()url="https://api.dingtalk.com/v1.0/calendar/users/me/events"headers={"x-acs-dingtalk-access-token":token,"Content-Type":"application/json"}body={"summary":summary,"start":{"dateTime":start_time,"timeZone":"Asia/Shanghai"},"end":{"dateTime":end_time,"timeZone":"Asia/Shanghai"},"location":{"displayName":location}iflocationelse{},"description":description}resp=requests.post(url,headers=headers,json=body)data=resp.json()ifresp.status_code==200and"id"indata:returnf"已创建日程「{summary}」，时间{start_time}~{end_time}"else:returnf"创建日程失败:{data}"@tool(args_schema=QueryCalendarInput)defquery_calendar(start_time:str,end_time:str)->str:"""查询指定时间段内的钉钉日程。"""token=ding_client.get_access_token()url="https://api.dingtalk.com/v1.0/calendar/users/me/events"params={"startTime":start_time,"endTime":end_time}headers={"x-acs-dingtalk-access-token":token}resp=requests.get(url,headers=headers,params=params)data=resp.json()events=data.get("events",[])ifnotevents:return"该时段暂无日程。"result=[]foreinevents:result.append(f"-{e['summary']}({e['start']['dateTime']}~{e['end']['dateTime']})")return"\n".join(result)@tool(args_schema=ModifyCalendarInput)defmodify_calendar_event(event_id:str,summary:str=None,start_time:str=None,end_time:str=None)->str:"""修改指定日程的标题或时间。"""token=ding_client.get_access_token()url=f"https://api.dingtalk.com/v1.0/calendar/users/me/events/{event_id}"headers={"x-acs-dingtalk-access-token":token,"Content-Type":"application/json"}body={}ifsummary:body["summary"]=summaryifstart_time:body["start"]={"dateTime":start_time,"timeZone":"Asia/Shanghai"}ifend_time:body["end"]={"dateTime":end_time,"timeZone":"Asia/Shanghai"}resp=requests.put(url,headers=headers,json=body)ifresp.status_code==200:returnf"日程{event_id}已更新。"else:returnf"更新失败:{resp.json()}"@tool(args_schema=DeleteCalendarInput)defdelete_calendar_event(event_id:str)->str:"""删除指定的钉钉日程。"""token=ding_client.get_access_token()url=f"https://api.dingtalk.com/v1.0/calendar/users/me/events/{event_id}"headers={"x-acs-dingtalk-access-token":token}resp=requests.delete(url,headers=headers)ifresp.status_code==200:returnf"日程{event_id}已删除。"else:returnf"删除失败:{resp.json()}"# 汇总工具列表ALL_TOOLS=[web_search,search_knowledge_base,create_todo,create_calendar_event,query_calendar,modify_calendar_event,delete_calendar_event,]

完整可运行的演示main.py

将上述工具与一个简单的 Agent 结合，测试工具调用效果（需本地准备.env和向量数据库，若无知识库可暂时注释掉知识库工具）。

# main.pyfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromconfigimportConfigfromtoolsimportALL_TOOLSdefmain():llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0,openai_api_key=Config.OPENAI_API_KEY,base_url=Config.OPENAI_BASE_URL,)prompt=ChatPromptTemplate.from_messages([("system","你是一个智能助手，可以使用工具帮助用户。"),MessagesPlaceholder("chat_history"),("human","{input}"),MessagesPlaceholder("agent_scratchpad")])agent=create_tool_calling_agent(llm,ALL_TOOLS,prompt)agent_executor=AgentExecutor(agent=agent,tools=ALL_TOOLS,verbose=True,handle_parsing_errors=True,)# 测试搜索print("--- 测试搜索 ---")res=agent_executor.invoke({"input":"今天的科技头条有哪些？"})print("回复:",res["output"][:200])# 测试待办（模拟负面情绪触发）print("\n--- 测试待办创建 ---")res=agent_executor.invoke({"input":"我非常生气！你们的产品根本用不了，立刻给我处理！"})print("回复:",res["output"])if__name__=="__main__":main()

运行说明：

1. 在.env中填入OPENAI_API_KEY、SERPAPI_API_KEY、钉钉应用凭证等。
2. 若要使用知识库，需提前运行嵌入脚本创建chroma_db目录（此处省略，可单独编写）。
3. 执行python main.py即可看到 Agent 自动选择并调用工具。

设计思想回顾

上述实现完美体现了文章开头强调的原则：

• 最小颗粒度：日历的增删查改被拆分为四个独立函数，智能体根据需求自行组合，例如“把明天所有会议推迟一小时”会先查询，再逐一修改。
• 结构化输入：所有工具的参数都通过args_schema绑定Pydantic模型，大模型传参的错误率大大降低。
• 工具描述即文档：每个@tool函数的 docstring 会被转换为工具描述，成为模型选用的唯一参考，必须精确无歧义。

工具开发的精髓不在于堆砌数量，而在于设计出能被智能体高效调用的接口。当你掌握了这些模式，就可以将任何外部系统（CRM、ERP、IoT）变成 Agent 的延伸。