AI Agent白手起家54: 工具设计原则与实现——搜索、知识库与钉钉API集成

📅 发布时间:2026/8/10 21:34:09
AI Agent白手起家54: 工具设计原则与实现——搜索、知识库与钉钉API集成 纲要工具在智能体中的核心地位设计原则最小颗粒度与解耦结构化参数Pydantic模型绑定工具描述与tool装饰器四大工具实现详解在线搜索SerpAPI封装知识库检索Chroma向量库 查询重写钉钉待办情感阈值触发创建钉钉日历增删查改功能钉钉 API 客户端封装完整可运行示例总结与相关度说明工具智能体的能力边界如果说大模型是智能体的大脑那么工具就是它的双手。一个智能体能做什么、做得多好几乎完全取决于所集成的工具。在 LangChain 中工具可以是一个函数、一个 API 调用、一个数据库查询——本质上任何能够接收输入并返回结果的外部能力都可以封装为工具。构建工具时有三个核心原则值得反复思考最小颗粒度将功能拆分为原子操作让智能体自己组合而不是代劳组合逻辑。结构化输入强制工具的入参拥有明确的类型和描述减少大模型传参随机性导致的错误。清晰的功能描述每个工具必须附带精确的注释这些注释将被注入提示词成为模型选择工具的唯一依据。工具集概览小浪助手作为钉钉智能客服集成了四类工具工具名称功能关键技术web_search在线搜索实时信息SerpAPIWrappersearch_knowledge_base检索私有知识库Chroma向量数据库 多查询重写create_todo创建钉钉待办事项钉钉 API Pydantic结构化参数manage_calendar增删查改钉钉日程钉钉 API 四个独立子工具工具的实现环境准备依赖安装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_KEYos.getenv(OPENAI_API_KEY)OPENAI_BASE_URLos.getenv(OPENAI_BASE_URL,https://api.openai.com/v1)SERPAPI_API_KEYos.getenv(SERPAPI_API_KEY)DINGTALK_APP_KEYos.getenv(DINGTALK_APP_KEY)DINGTALK_APP_SECRETos.getenv(DINGTALK_APP_SECRET)DINGTALK_AGENT_IDos.getenv(DINGTALK_AGENT_ID)CHROMA_PERSIST_DIR./chroma_db钉钉客户端封装dingtalk_client.py钉钉 API 大多需要access_token我们封装一个简单的客户端负责获取和缓存。# dingtalk_client.pyimportrequestsimporttimefromconfigimportConfigclassDingTalkClient:def__init__(self):self.app_keyConfig.DINGTALK_APP_KEY self.app_secretConfig.DINGTALK_APP_SECRET self.agent_idConfig.DINGTALK_AGENT_ID self._tokenNoneself._expire_time0defget_access_token(self)-str:ifself._tokenandtime.time()self._expire_time:returnself._token urlhttps://oapi.dingtalk.com/gettokenparams{appkey:self.app_key,appsecret:self.app_secret}resprequests.get(url,paramsparams)dataresp.json()ifdata.get(errcode)0:self._tokendata[access_token]self._expire_timetime.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:strField(description待办事项标题)description:Optional[str]Field(None,description详细描述)priority:intField(20,description优先级数字越小越优先默认20)classCalendarEventInput(BaseModel):summary:strField(description日程标题)start_time:strField(description开始时间ISO格式如2025-01-01T10:00:0008:00)end_time:strField(description结束时间ISO格式)location:Optional[str]Field(None,description地点)description:Optional[str]Field(None,description日程描述)classQueryCalendarInput(BaseModel):start_time:strField(description查询时间段的开始ISO格式)end_time:strField(description查询时间段的结束ISO格式)classModifyCalendarInput(BaseModel):event_id:strField(description要修改的日程ID)summary:Optional[str]Field(None)start_time:Optional[str]Field(None)end_time:Optional[str]Field(None)classDeleteCalendarInput(BaseModel):event_id:strField(description要删除的日程ID)# ---------- 搜索工具 ----------tooldefweb_search(query:str)-str:在线搜索最新信息。当需要实时数据或未知事实时使用此工具。searchSerpAPIWrapper(serpapi_api_keyConfig.SERPAPI_API_KEY)returnsearch.run(query)# ---------- 知识库工具 ----------# 假设向量库已预先创建并填充这里仅演示检索embeddingsOpenAIEmbeddings(openai_api_keyConfig.OPENAI_API_KEY,base_urlConfig.OPENAI_BASE_URL)vectorstoreChroma(persist_directoryConfig.CHROMA_PERSIST_DIR,embedding_functionembeddings)defrewrite_query(original:str)-List[str]:查询重写生成多个变体以提高召回率llmChatOpenAI(modelgpt-3.5-turbo,temperature0.3)promptChatPromptTemplate.from_template(将以下用户问题改写为3个不同角度但语义相同的查询每个查询单独一行不要编号。\n问题: {query}\n改写的查询:)chainprompt|llm resultchain.invoke({query:original})queries[q.strip()forqinresult.content.split(\n)ifq.strip()]return[original]queriestooldefsearch_knowledge_base(query:str)-str:查询内部知识库获取LangChain等专业知识。queriesrewrite_query(query)all_docs[]forqinqueries:docsvectorstore.similarity_search(q,k2)all_docs.extend(docs)# 去重并取前5个最相关的unique_contents[]seenset()fordocinall_docs:ifdoc.page_contentnotinseen:seen.add(doc.page_content)unique_contents.append(doc.page_content)return\n\n.join(unique_contents[:5])# ---------- 钉钉待办工具 ----------ding_clientDingTalkClient()tool(args_schemaTodoInput)defcreate_todo(subject:str,description:str,priority:int20)-str:创建钉钉待办事项。当用户要求记录事务、投诉或强烈负面情绪时使用。tokending_client.get_access_token()urlhttps://api.dingtalk.com/v1.0/todo/users/me/tasksheaders{x-acs-dingtalk-access-token:token,Content-Type:application/json}body{subject:subject,description:description,priority:priority,createdTime:int(datetime.now().timestamp()*1000)}resprequests.post(url,headersheaders,jsonbody)dataresp.json()ifresp.status_code200andidindata:returnf已创建待办「{subject}」优先级{priority}else:returnf创建待办失败:{data}# ---------- 钉钉日历工具 ----------tool(args_schemaCalendarEventInput)defcreate_calendar_event(summary:str,start_time:str,end_time:str,location:str,description:str)-str:在钉钉日历中创建新日程。参数均为ISO 8601格式字符串。tokending_client.get_access_token()urlhttps://api.dingtalk.com/v1.0/calendar/users/me/eventsheaders{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}resprequests.post(url,headersheaders,jsonbody)dataresp.json()ifresp.status_code200andidindata:returnf已创建日程「{summary}」时间{start_time}~{end_time}else:returnf创建日程失败:{data}tool(args_schemaQueryCalendarInput)defquery_calendar(start_time:str,end_time:str)-str:查询指定时间段内的钉钉日程。tokending_client.get_access_token()urlhttps://api.dingtalk.com/v1.0/calendar/users/me/eventsparams{startTime:start_time,endTime:end_time}headers{x-acs-dingtalk-access-token:token}resprequests.get(url,headersheaders,paramsparams)dataresp.json()eventsdata.get(events,[])ifnotevents:return该时段暂无日程。result[]foreinevents:result.append(f-{e[summary]}({e[start][dateTime]}~{e[end][dateTime]}))return\n.join(result)tool(args_schemaModifyCalendarInput)defmodify_calendar_event(event_id:str,summary:strNone,start_time:strNone,end_time:strNone)-str:修改指定日程的标题或时间。tokending_client.get_access_token()urlfhttps://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}resprequests.put(url,headersheaders,jsonbody)ifresp.status_code200:returnf日程{event_id}已更新。else:returnf更新失败:{resp.json()}tool(args_schemaDeleteCalendarInput)defdelete_calendar_event(event_id:str)-str:删除指定的钉钉日程。tokending_client.get_access_token()urlfhttps://api.dingtalk.com/v1.0/calendar/users/me/events/{event_id}headers{x-acs-dingtalk-access-token:token}resprequests.delete(url,headersheaders)ifresp.status_code200: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():llmChatOpenAI(modelgpt-3.5-turbo,temperature0,openai_api_keyConfig.OPENAI_API_KEY,base_urlConfig.OPENAI_BASE_URL,)promptChatPromptTemplate.from_messages([(system,你是一个智能助手可以使用工具帮助用户。),MessagesPlaceholder(chat_history),(human,{input}),MessagesPlaceholder(agent_scratchpad)])agentcreate_tool_calling_agent(llm,ALL_TOOLS,prompt)agent_executorAgentExecutor(agentagent,toolsALL_TOOLS,verboseTrue,handle_parsing_errorsTrue,)# 测试搜索print(--- 测试搜索 ---)resagent_executor.invoke({input:今天的科技头条有哪些})print(回复:,res[output][:200])# 测试待办模拟负面情绪触发print(\n--- 测试待办创建 ---)resagent_executor.invoke({input:我非常生气你们的产品根本用不了立刻给我处理})print(回复:,res[output])if__name____main__:main()运行说明在.env中填入OPENAI_API_KEY、SERPAPI_API_KEY、钉钉应用凭证等。若要使用知识库需提前运行嵌入脚本创建chroma_db目录此处省略可单独编写。执行python main.py即可看到 Agent 自动选择并调用工具。设计思想回顾上述实现完美体现了文章开头强调的原则最小颗粒度日历的增删查改被拆分为四个独立函数智能体根据需求自行组合例如“把明天所有会议推迟一小时”会先查询再逐一修改。结构化输入所有工具的参数都通过args_schema绑定Pydantic模型大模型传参的错误率大大降低。工具描述即文档每个tool函数的 docstring 会被转换为工具描述成为模型选用的唯一参考必须精确无歧义。工具开发的精髓不在于堆砌数量而在于设计出能被智能体高效调用的接口。当你掌握了这些模式就可以将任何外部系统CRM、ERP、IoT变成 Agent 的延伸。