Flink Agents 與 AI 流式應(yīng)用 —— 第 4 章 Tool 機(jī)制詳解讓 Agent 能動(dòng)手章節(jié)定位第二篇開篇。沒有工具的 Agent 只是聊天機(jī)器人——本章把 Agent 的手做扎實(shí)Tool 定義三要素的生產(chǎn)級(jí)寫法、Function Calling 全鏈路拆解、同步/異步/流式三類工具開發(fā)、超時(shí)/冪等/錯(cuò)誤碼語義化的工程化要點(diǎn)實(shí)戰(zhàn)把車聯(lián)網(wǎng)四大系統(tǒng)封裝成工具。學(xué)完本章你的 Agent 能查檔案、跑 SQL、建工單、下指令。版本基線Apache Flink 2.3.0 Flink Agents 0.3.xJDK 17章節(jié)導(dǎo)讀4.1 Tool 定義三要素生產(chǎn)級(jí)寫法4.2 Function Calling 全鏈路拆解4.3 自定義 Tool 開發(fā)同步 / 異步 / 流式4.4 工具工程化超時(shí)、冪等與錯(cuò)誤碼語義化4.5 實(shí)戰(zhàn)車聯(lián)網(wǎng)四大系統(tǒng)工具化4.6 本章小結(jié)與下章預(yù)告4.1 Tool 定義三要素生產(chǎn)級(jí)寫法4.1.1 名稱與描述LLM 選型的唯一依據(jù)LLM 看不到你的代碼,只看到工具清單: ? 差描述:查歷史 → 模型不知道什么時(shí)候該用它,參數(shù)全靠猜 ? 好描述:查詢指定車輛近 N 個(gè)月的歷史維修工單, 返回故障碼/處置方案/復(fù)檢結(jié)果。當(dāng)告警診斷需要 歷史故障佐證時(shí)使用。vin 為 17 位大寫車架號(hào)。 → 用途、時(shí)機(jī)、參數(shù)格式、返回內(nèi)容,四個(gè)信息齊全描述四要素干什么、什么時(shí)候用、參數(shù)怎么填、返回什么。描述是 prompt 的一部分寫好描述比調(diào)參收益大。4.1.2 參數(shù) schema框架校驗(yàn)的契約ToolcreateTicketTool.builder().name(create_work_ticket).description(創(chuàng)建維修工單。僅當(dāng)診斷結(jié)論明確且置信度≥0.8時(shí)使用。返回工單號(hào)。重復(fù)創(chuàng)建同 event_id 工單會(huì)自動(dòng)去重。).paramSchema(Map.of(event_id,Param.string(關(guān)聯(lián)告警 ID,冪等鍵,true),vin,Param.string(17 位車架號(hào),true),level,Param.enumeration(工單等級(jí),List.of(P0,P1,P2,P3),true),plan,Param.string(處置方案,200 字內(nèi),true))).executor(ticketExecutor).build();schema 三原則: ① 枚舉盡量封閉:level 枚舉四值,不給模型自由發(fā)揮空間(前作 6.5 思想) ② 必填最小化:可選參數(shù)給默認(rèn)值,降低模型生成負(fù)擔(dān) ③ 冪等鍵顯式化:event_id 寫進(jìn)描述,讓模型知道必須帶上4.1.3 返回值契約返回結(jié)構(gòu)化 JSON,而非自然語言: {ticket_no:WO20261010001,status:CREATED,deduplicated:false} 為什么: ① LLM 解析結(jié)構(gòu)化結(jié)果比解析散文可靠 ② 框架可對(duì)返回值做二次校驗(yàn) ③ 審計(jì)日志直接可用(第 14 章)4.2 Function Calling 全鏈路拆解┌────────────────────────────────────────────────────────┐ │ ① 裝配:Agent 啟動(dòng)時(shí),工具清單(名稱描述schema) │ │ 注入系統(tǒng) prompt 或走模型原生 function-calling 通道 │ │ │ │ ② 決策:LLM 輸出結(jié)構(gòu)化調(diào)用意圖 │ │ {tool:query_vehicle_history,args:{vin:...}} │ │ │ │ ③ 校驗(yàn):框架按 schema 檢查 │ │ 缺必填 → 生成錯(cuò)誤提示回喂 LLM 重生成(不重試工具) │ │ 類型錯(cuò) → 同上 │ │ 校驗(yàn)通過 → 進(jìn)入執(zhí)行 │ │ │ │ ④ 執(zhí)行:ToolRequestEvent → 執(zhí)行器(同步/異步) │ │ 超時(shí)/異常 → 錯(cuò)誤語義化后作為結(jié)果回填(見 4.4.3) │ │ │ │ ⑤ 回填:ToolResponseEvent 帶結(jié)果回到上下文 │ │ LLM 基于結(jié)果繼續(xù)推理(可能再調(diào)下一個(gè)工具) │ └────────────────────────────────────────────────────────┘關(guān)鍵點(diǎn)校驗(yàn)失敗回喂 LLM而不是拋異常。模型看到缺少必填參數(shù) vin會(huì)自我修正一次重生成遠(yuǎn)比作業(yè)失敗重啟便宜。4.3 自定義 Tool 開發(fā)同步 / 異步 / 流式4.3.1 同步工具簡(jiǎn)單查詢// 適合:毫秒級(jí)返回的本地/內(nèi)存操作.executor((args)-faultCodeDict.lookup((String)args.get(code)))4.3.2 異步工具外部 API生產(chǎn)主力// 適合:網(wǎng)絡(luò)調(diào)用,不阻塞事件循環(huán)(前作 12 章異步思想延續(xù)).executorAsync((args)-CompletableFuture.supplyAsync(()-dorisClient.queryWorkOrders((String)args.get(vin)),toolExecutorPool)// 獨(dú)立線程池,與算子線程隔離.orTimeout(10,TimeUnit.SECONDS))// 工具級(jí)超時(shí)線程池紀(jì)律: 工具池與 Flink 算子線程隔離——慢工具不拖垮事件循環(huán) 池大小按工具 QPS×延遲測(cè)算(前作 12.6 公式)4.3.3 流式工具大結(jié)果集分批返回場(chǎng)景:召回 50 條相似工單,逐批回填,LLM 邊收邊讀 適用:結(jié)果 10KB 或生成時(shí)間長(zhǎng)的工具 注意:流式結(jié)果要有明確結(jié)束標(biāo)記,框架才知道何時(shí)喚醒后續(xù) Action4.3.4 選型速查類型延遲畫像典型工具同步100ms字典查詢、本地計(jì)算、緩存讀異步100ms~30sDoris 查詢、業(yè)務(wù) API、工單創(chuàng)建流式長(zhǎng)結(jié)果/生成式大批量召回、報(bào)告生成4.4 工具工程化超時(shí)、冪等與錯(cuò)誤碼語義化4.4.1 超時(shí)雙層防護(hù)工具級(jí)超時(shí):executor 內(nèi) orTimeout(10s)——快失敗 框架級(jí)超時(shí):ToolRequestEvent 攜帶 deadline——兜底 兩層都要:工具級(jí)保護(hù)線程池,框架級(jí)保護(hù)事件循環(huán) 超時(shí)結(jié)果 → 語義化錯(cuò)誤回填(不是異常炸作業(yè))4.4.2 冪等動(dòng)作類工具的紅線動(dòng)作類工具(建單/下發(fā)指令/寫庫)必須冪等: ① 冪等鍵:event_id 作為業(yè)務(wù)去重鍵 ② 執(zhí)行前去重:查該 event_id 是否已建單 ③ 返回去重標(biāo)記:{deduplicated:true,ticket_no:原單號(hào)} 為什么必須: Agent 決策可能因 Checkpoint 重放而重復(fù)執(zhí)行(13.2 詳述) 不冪等 故障恢復(fù)后重復(fù)建單、重復(fù)下發(fā)指令——生產(chǎn)事故4.4.3 錯(cuò)誤碼語義化讓 LLM 讀懂失敗? 拋異常:java.sql.SQLTimeoutException → LLM 看不懂,可能反復(fù)重試同一個(gè)錯(cuò) ? 語義化返回: {error:QUERY_TIMEOUT,message:查詢超時(shí),可能是時(shí)間范圍過大, retryable:true,suggestion:縮小 months 參數(shù)至 3 后重試} LLM 看到后會(huì)自我修正:縮小時(shí)間范圍重查—— 錯(cuò)誤信息是 Agent 自愈的教材,寫它就是寫 prompt4.5 實(shí)戰(zhàn)車聯(lián)網(wǎng)四大系統(tǒng)工具化4.5.1 工具清單工具類型系統(tǒng)動(dòng)作級(jí)別query_vehicle_history異步Doris 工單庫只讀query_vehicle_profile異步Doris 檔案只讀run_adhoc_sql異步Flink SQL 網(wǎng)關(guān)只讀白名單create_work_ticket異步工單系統(tǒng)寫冪等審批策略query_similar_cases異步Doris ANN前作 RAG只讀send_remote_diagnose異步TSP 遠(yuǎn)程診斷高危強(qiáng)制審批4.5.2 相似案例檢索工具復(fù)用前作 RAG 資產(chǎn)ToolquerySimilarCasesTool.builder().name(query_similar_cases).description(語義檢索相似歷史故障案例。返回 Top5 案例的故障現(xiàn)象/根因/處置方案/復(fù)檢結(jié)果。診斷拿不準(zhǔn)時(shí)使用。).paramSchema(Map.of(symptom,Param.string(故障現(xiàn)象描述,true),series,Param.string(車系,如 EV-X9,true))).executorAsync((args)-CompletableFuture.supplyAsync(()-{// ① 調(diào) embedding 模型向量化 symptom(前作第 7 章資產(chǎn))float[]vecembedClient.embed((String)args.get(symptom));// ② Doris ANN 混合查詢:標(biāo)量過濾車系 向量召回(前作 15.4 資產(chǎn))returndorisClient.annSearch(ai.doc_vector,vec,series args.get(series),5);},toolExecutorPool).orTimeout(15,TimeUnit.SECONDS)).build();前作第 18 章建的維修知識(shí)庫本章變成 Agent 手里的一個(gè)工具——資產(chǎn)復(fù)用的典型案例。4.5.3 高危工具的安全殼ToolremoteDiagnoseTool.builder().name(send_remote_diagnose).description(向車輛下發(fā)遠(yuǎn)程診斷指令(讀取 ECU 數(shù)據(jù))。高危操作,必須先在描述中包含 approval_id(人工審批號(hào))。).paramSchema(Map.of(vin,Param.string(車架號(hào),true),command,Param.enumeration(指令,List.of(READ_ECU,READ_BMS_DETAIL),true),approval_id,Param.string(人工審批號(hào),無審批禁止調(diào)用,true))).executorAsync((args)-{// 雙保險(xiǎn):代碼層再校驗(yàn)審批號(hào)真實(shí)存在且未使用if(!approvalService.verify((String)args.get(approval_id))){returnCompletableFuture.completedFuture(Map.of(error,APPROVAL_INVALID,message,審批號(hào)無效,禁止執(zhí)行。請(qǐng)先發(fā)起審批。,retryable,false));}returntspClient.sendDiagnose(...);}).build();高危工具三保險(xiǎn)(第 15 章全面展開): ① prompt 約束:系統(tǒng)提示中聲明無 approval_id 禁止調(diào)用 ② schema 約束:approval_id 必填 ③ 代碼校驗(yàn):審批號(hào)真實(shí)性服務(wù)端再驗(yàn)——不信任模型自覺4.5.4 工具注冊(cè)與驗(yàn)證AgentagentnewAlertAgent(model);agent.registerTools(List.of(queryHistory,queryProfile,querySimilarCases,// 只讀三件套createTicket,// 寫,冪等remoteDiagnose// 高危,審批));// 上線前驗(yàn)證清單:// □ 每個(gè)工具描述含四要素 □ 動(dòng)作類全部?jī)绲?□ 高危類全部三保險(xiǎn)// □ 全部工具有超時(shí) □ 錯(cuò)誤返回全部語義化4.6 本章小結(jié)與下章預(yù)告本章小結(jié)┌──────────────────────────────────────────────────────────────┐ │ 第 4 章 要點(diǎn)回顧 │ └──────────────────────────────────────────────────────────────┘ ? 三要素:描述四要素(干什么/何時(shí)用/怎么填/返回什么); schema 枚舉封閉必填最小化冪等鍵顯式;返回結(jié)構(gòu)化 ? 全鏈路:裝配→決策→校驗(yàn)→執(zhí)行→回填; 校驗(yàn)失敗回喂 LLM 重生成,不拋異常 ? 三類型:同步(100ms)/異步(網(wǎng)絡(luò)主力,獨(dú)立線程池)/ 流式(大結(jié)果分批) ? 工程化:雙層超時(shí);動(dòng)作類冪等(event_id 去重); 錯(cuò)誤語義化Agent 自愈的教材 ? 實(shí)戰(zhàn):六工具清單;RAG 資產(chǎn)變工具;高危三保險(xiǎn) (promptschema代碼校驗(yàn))下章預(yù)告第 5 章 MCP 集成標(biāo)準(zhǔn)化工具生態(tài)手工作坊升級(jí)到標(biāo)準(zhǔn)生態(tài)——MCP 協(xié)議解析、Flink Agents 作為 MCP Client 的發(fā)現(xiàn)/連接/調(diào)用、生態(tài) Server 速覽、自建 MCP Server 把 TSP/CRM/工單系統(tǒng)標(biāo)準(zhǔn)化暴露、MCP 鑒權(quán)與審計(jì)實(shí)戰(zhàn)通過 MCP 接入故障碼庫與前作知識(shí)庫 RAG。官方參考資料Flink Agents 0.3 DocumentationTool 與 MCP 章節(jié)Flink Agents 0.3.0 Release AnnouncementModel Context Protocol 官方文檔本章實(shí)戰(zhàn)代碼清單序號(hào)內(nèi)容說明1工具描述四要素正反對(duì)照4.1.12建單工具完整定義枚舉冪等鍵4.1.23Function Calling 五步鏈路圖4.24異步工具開發(fā)模板獨(dú)立線程池超時(shí)4.3.25錯(cuò)誤語義化返回規(guī)范4.4.36相似案例檢索工具復(fù)用 RAG 資產(chǎn)4.5.27高危工具三保險(xiǎn)實(shí)現(xiàn)4.5.3