Flink Agents 与 AI 流式应用 —— 第 4 章 Tool 机制详解让 Agent 能动手章节定位第二篇开篇。没有工具的 Agent 只是聊天机器人——本章把 Agent 的手做扎实Tool 定义三要素的生产级写法、Function Calling 全链路拆解、同步/异步/流式三类工具开发、超时/幂等/错误码语义化的工程化要点实战把车联网四大系统封装成工具。学完本章你的 Agent 能查档案、跑 SQL、建工单、下指令。版本基线Apache Flink 2.3.0 Flink Agents 0.3.xJDK 17章节导读4.1 Tool 定义三要素生产级写法4.2 Function Calling 全链路拆解4.3 自定义 Tool 开发同步 / 异步 / 流式4.4 工具工程化超时、幂等与错误码语义化4.5 实战车联网四大系统工具化4.6 本章小结与下章预告4.1 Tool 定义三要素生产级写法4.1.1 名称与描述LLM 选型的唯一依据LLM 看不到你的代码,只看到工具清单: ❌ 差描述:查历史 → 模型不知道什么时候该用它,参数全靠猜 ✅ 好描述:查询指定车辆近 N 个月的历史维修工单, 返回故障码/处置方案/复检结果。当告警诊断需要 历史故障佐证时使用。vin 为 17 位大写车架号。 → 用途、时机、参数格式、返回内容,四个信息齐全描述四要素干什么、什么时候用、参数怎么填、返回什么。描述是 prompt 的一部分写好描述比调参收益大。4.1.2 参数 schema框架校验的契约ToolcreateTicketTool.builder().name(create_work_ticket).description(创建维修工单。仅当诊断结论明确且置信度≥0.8时使用。返回工单号。重复创建同 event_id 工单会自动去重。).paramSchema(Map.of(event_id,Param.string(关联告警 ID,幂等键,true),vin,Param.string(17 位车架号,true),level,Param.enumeration(工单等级,List.of(P0,P1,P2,P3),true),plan,Param.string(处置方案,200 字内,true))).executor(ticketExecutor).build();schema 三原则: ① 枚举尽量封闭:level 枚举四值,不给模型自由发挥空间(前作 6.5 思想) ② 必填最小化:可选参数给默认值,降低模型生成负担 ③ 幂等键显式化:event_id 写进描述,让模型知道必须带上4.1.3 返回值契约返回结构化 JSON,而非自然语言: {ticket_no:WO20261010001,status:CREATED,deduplicated:false} 为什么: ① LLM 解析结构化结果比解析散文可靠 ② 框架可对返回值做二次校验 ③ 审计日志直接可用(第 14 章)4.2 Function Calling 全链路拆解┌────────────────────────────────────────────────────────┐ │ ① 装配:Agent 启动时,工具清单(名称描述schema) │ │ 注入系统 prompt 或走模型原生 function-calling 通道 │ │ │ │ ② 决策:LLM 输出结构化调用意图 │ │ {tool:query_vehicle_history,args:{vin:...}} │ │ │ │ ③ 校验:框架按 schema 检查 │ │ 缺必填 → 生成错误提示回喂 LLM 重生成(不重试工具) │ │ 类型错 → 同上 │ │ 校验通过 → 进入执行 │ │ │ │ ④ 执行:ToolRequestEvent → 执行器(同步/异步) │ │ 超时/异常 → 错误语义化后作为结果回填(见 4.4.3) │ │ │ │ ⑤ 回填:ToolResponseEvent 带结果回到上下文 │ │ LLM 基于结果继续推理(可能再调下一个工具) │ └────────────────────────────────────────────────────────┘关键点校验失败回喂 LLM而不是抛异常。模型看到缺少必填参数 vin会自我修正一次重生成远比作业失败重启便宜。4.3 自定义 Tool 开发同步 / 异步 / 流式4.3.1 同步工具简单查询// 适合:毫秒级返回的本地/内存操作.executor((args)-faultCodeDict.lookup((String)args.get(code)))4.3.2 异步工具外部 API生产主力// 适合:网络调用,不阻塞事件循环(前作 12 章异步思想延续).executorAsync((args)-CompletableFuture.supplyAsync(()-dorisClient.queryWorkOrders((String)args.get(vin)),toolExecutorPool)// 独立线程池,与算子线程隔离.orTimeout(10,TimeUnit.SECONDS))// 工具级超时线程池纪律: 工具池与 Flink 算子线程隔离——慢工具不拖垮事件循环 池大小按工具 QPS×延迟测算(前作 12.6 公式)4.3.3 流式工具大结果集分批返回场景:召回 50 条相似工单,逐批回填,LLM 边收边读 适用:结果 10KB 或生成时间长的工具 注意:流式结果要有明确结束标记,框架才知道何时唤醒后续 Action4.3.4 选型速查类型延迟画像典型工具同步100ms字典查询、本地计算、缓存读异步100ms~30sDoris 查询、业务 API、工单创建流式长结果/生成式大批量召回、报告生成4.4 工具工程化超时、幂等与错误码语义化4.4.1 超时双层防护工具级超时:executor 内 orTimeout(10s)——快失败 框架级超时:ToolRequestEvent 携带 deadline——兜底 两层都要:工具级保护线程池,框架级保护事件循环 超时结果 → 语义化错误回填(不是异常炸作业)4.4.2 幂等动作类工具的红线动作类工具(建单/下发指令/写库)必须幂等: ① 幂等键:event_id 作为业务去重键 ② 执行前去重:查该 event_id 是否已建单 ③ 返回去重标记:{deduplicated:true,ticket_no:原单号} 为什么必须: Agent 决策可能因 Checkpoint 重放而重复执行(13.2 详述) 不幂等 故障恢复后重复建单、重复下发指令——生产事故4.4.3 错误码语义化让 LLM 读懂失败❌ 抛异常:java.sql.SQLTimeoutException → LLM 看不懂,可能反复重试同一个错 ✅ 语义化返回: {error:QUERY_TIMEOUT,message:查询超时,可能是时间范围过大, retryable:true,suggestion:缩小 months 参数至 3 后重试} LLM 看到后会自我修正:缩小时间范围重查—— 错误信息是 Agent 自愈的教材,写它就是写 prompt4.5 实战车联网四大系统工具化4.5.1 工具清单工具类型系统动作级别query_vehicle_history异步Doris 工单库只读query_vehicle_profile异步Doris 档案只读run_adhoc_sql异步Flink SQL 网关只读白名单create_work_ticket异步工单系统写幂等审批策略query_similar_cases异步Doris ANN前作 RAG只读send_remote_diagnose异步TSP 远程诊断高危强制审批4.5.2 相似案例检索工具复用前作 RAG 资产ToolquerySimilarCasesTool.builder().name(query_similar_cases).description(语义检索相似历史故障案例。返回 Top5 案例的故障现象/根因/处置方案/复检结果。诊断拿不准时使用。).paramSchema(Map.of(symptom,Param.string(故障现象描述,true),series,Param.string(车系,如 EV-X9,true))).executorAsync((args)-CompletableFuture.supplyAsync(()-{// ① 调 embedding 模型向量化 symptom(前作第 7 章资产)float[]vecembedClient.embed((String)args.get(symptom));// ② Doris ANN 混合查询:标量过滤车系 向量召回(前作 15.4 资产)returndorisClient.annSearch(ai.doc_vector,vec,series args.get(series),5);},toolExecutorPool).orTimeout(15,TimeUnit.SECONDS)).build();前作第 18 章建的维修知识库本章变成 Agent 手里的一个工具——资产复用的典型案例。4.5.3 高危工具的安全壳ToolremoteDiagnoseTool.builder().name(send_remote_diagnose).description(向车辆下发远程诊断指令(读取 ECU 数据)。高危操作,必须先在描述中包含 approval_id(人工审批号)。).paramSchema(Map.of(vin,Param.string(车架号,true),command,Param.enumeration(指令,List.of(READ_ECU,READ_BMS_DETAIL),true),approval_id,Param.string(人工审批号,无审批禁止调用,true))).executorAsync((args)-{// 双保险:代码层再校验审批号真实存在且未使用if(!approvalService.verify((String)args.get(approval_id))){returnCompletableFuture.completedFuture(Map.of(error,APPROVAL_INVALID,message,审批号无效,禁止执行。请先发起审批。,retryable,false));}returntspClient.sendDiagnose(...);}).build();高危工具三保险(第 15 章全面展开): ① prompt 约束:系统提示中声明无 approval_id 禁止调用 ② schema 约束:approval_id 必填 ③ 代码校验:审批号真实性服务端再验——不信任模型自觉4.5.4 工具注册与验证AgentagentnewAlertAgent(model);agent.registerTools(List.of(queryHistory,queryProfile,querySimilarCases,// 只读三件套createTicket,// 写,幂等remoteDiagnose// 高危,审批));// 上线前验证清单:// □ 每个工具描述含四要素 □ 动作类全部幂等 □ 高危类全部三保险// □ 全部工具有超时 □ 错误返回全部语义化4.6 本章小结与下章预告本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 4 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 三要素:描述四要素(干什么/何时用/怎么填/返回什么); schema 枚举封闭必填最小化幂等键显式;返回结构化 ✓ 全链路:装配→决策→校验→执行→回填; 校验失败回喂 LLM 重生成,不抛异常 ✓ 三类型:同步(100ms)/异步(网络主力,独立线程池)/ 流式(大结果分批) ✓ 工程化:双层超时;动作类幂等(event_id 去重); 错误语义化Agent 自愈的教材 ✓ 实战:六工具清单;RAG 资产变工具;高危三保险 (promptschema代码校验)下章预告第 5 章 MCP 集成标准化工具生态手工作坊升级到标准生态——MCP 协议解析、Flink Agents 作为 MCP Client 的发现/连接/调用、生态 Server 速览、自建 MCP Server 把 TSP/CRM/工单系统标准化暴露、MCP 鉴权与审计实战通过 MCP 接入故障码库与前作知识库 RAG。官方参考资料Flink Agents 0.3 DocumentationTool 与 MCP 章节Flink Agents 0.3.0 Release AnnouncementModel Context Protocol 官方文档本章实战代码清单序号内容说明1工具描述四要素正反对照4.1.12建单工具完整定义枚举幂等键4.1.23Function Calling 五步链路图4.24异步工具开发模板独立线程池超时4.3.25错误语义化返回规范4.4.36相似案例检索工具复用 RAG 资产4.5.27高危工具三保险实现4.5.3