企业官网建设流程全解析

Flink Agents 与 AI 流式应用 —— 第 4 章 Tool 机制详解：让 Agent 能动手

章节定位：第二篇开篇。没有工具的 Agent 只是聊天机器人——本章把 Agent 的"手"做扎实：Tool 定义三要素的生产级写法、Function Calling 全链路拆解、同步/异步/流式三类工具开发、超时/幂等/错误码语义化的工程化要点，实战把车联网四大系统封装成工具。学完本章，你的 Agent 能查档案、跑 SQL、建工单、下指令。

版本基线：Apache Flink 2.3.0 + Flink Agents 0.3.x（JDK 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：框架校验的契约

ToolcreateTicket=Tool.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 或生成时间长的工具 注意:流式结果要有明确结束标记,框架才知道何时唤醒后续 Action

4.3.4 选型速查

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 自愈的教材,写它就是写 prompt

4.5 实战：车联网四大系统工具化

4.5.1 工具清单

4.5.2 相似案例检索工具（复用前作 RAG 资产）

ToolquerySimilarCases=Tool.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[]vec=embedClient.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 高危工具的安全壳

ToolremoteDiagnose=Tool.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 工具注册与验证

Agentagent=newAlertAgent(model);agent.registerTools(List.of(queryHistory,queryProfile,querySimilarCases,// 只读三件套createTicket,// 写,幂等remoteDiagnose// 高危,审批));// 上线前验证清单:// □ 每个工具描述含四要素 □ 动作类全部幂等 □ 高危类全部三保险// □ 全部工具有超时 □ 错误返回全部语义化

4.6 本章小结与下章预告

本章小结

┌──────────────────────────────────────────────────────────────┐ │ 第 4 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 三要素:描述四要素(干什么/何时用/怎么填/返回什么); schema 枚举封闭+必填最小化+幂等键显式;返回结构化 ✓ 全链路:装配→决策→校验→执行→回填; 校验失败回喂 LLM 重生成,不抛异常 ✓ 三类型:同步(<100ms)/异步(网络主力,独立线程池)/ 流式(大结果分批) ✓ 工程化:双层超时;动作类幂等(event_id 去重); 错误语义化=Agent 自愈的教材 ✓ 实战:六工具清单;RAG 资产变工具;高危三保险 (prompt+schema+代码校验)

下章预告

第 5 章 MCP 集成：标准化工具生态

手工作坊升级到标准生态——MCP 协议解析、Flink Agents 作为 MCP Client 的发现/连接/调用、生态 Server 速览、自建 MCP Server 把 TSP/CRM/工单系统标准化暴露、MCP 鉴权与审计，实战通过 MCP 接入故障码库与前作知识库 RAG。

官方参考资料

• Flink Agents 0.3 Documentation（Tool 与 MCP 章节）
• Flink Agents 0.3.0 Release Announcement
• Model Context Protocol 官方文档

本章实战代码清单

序号	内容	说明
1	工具描述四要素正反对照	4.1.1
2	建单工具完整定义（枚举+幂等键）	4.1.2
3	Function Calling 五步链路图	4.2
4	异步工具开发模板（独立线程池+超时）	4.3.2
5	错误语义化返回规范	4.4.3
6	相似案例检索工具（复用 RAG 资产）	4.5.2
7	高危工具三保险实现	4.5.3