AI应用数据建模：Message与ToolCall设计及流式管道实践

1. 从一次“数据格式不兼容”的报错说起最近在调试一个基于大语言模型的应用时我又一次遇到了那个熟悉的错误data incompatible with messages format. each message should be a dictionary...。这个报错背后本质上是一个数据建模问题——我的程序期望的Message对象结构与实际传入的数据对不上。这让我想起了在设计和实现Eino一个假设的、用于构建复杂AI工作流的框架或工具这类系统时核心挑战之一就是如何为AI交互过程中的核心实体如Message、ToolCall设计一套清晰、健壮且可扩展的数据模型并在此基础上构建高效的流式处理管道。简单来说Eino的数据建模要解决的是“如何用代码精准地描述和操控一次AI对话”的问题。这不仅仅是定义几个类Class那么简单。它需要你思考用户的一句话、AI的一次回复、AI调用外部工具的一次请求及其结果这些信息应该如何被结构化地存储、传递和转换尤其是在追求流式Streaming体验的今天当AI的回复是一个字一个字“吐”出来的时候背后的数据管道又该如何设计才能既保证实时性又不丢失结构化的语义信息比如流到一半AI突然说要调用一个工具这个信号该如何捕获和处理如果你正在构建涉及复杂AI Agent、多轮工具调用或需要精细控制对话流程的应用那么理解Message、ToolCall的建模以及流式管道的设计将是绕不开的一课。这不仅能帮你避免开头提到的低级数据格式错误更能让你构建出响应迅速、逻辑清晰且易于维护的AI应用。接下来我将结合常见的实践模式深入拆解这几个核心概念。2. Message对话的基本单元及其Schema设计在任何对话式AI系统中Message都是最基础的构建块。它代表了一次单向的通信通常包含角色如user、assistant、system、内容以及可能的元数据。2.1 核心字段与角色定义一个最小化的Message模型通常包含以下字段role (string): 发送者角色。这是最关键的一个字段它决定了消息的语义。常见的角色有system: 系统指令用于设定AI的行为、人格或上下文规则。通常只在对话开始时出现一次。user: 用户输入的问题或指令。assistant: AI模型的回复。tool(或function): 代表工具调用的返回结果。这个角色对于实现工具调用至关重要。content (string 或 array): 消息的内容。对于简单文本它是一个字符串。但在更复杂的场景下如多模态输入或结构化内容它可能是一个由文本和图像URL等对象组成的数组。name (string, 可选): 当role为tool时通常用name来指定是哪个工具调用的返回结果。然而仅仅这样定义是远远不够的。在实际的Eino这类框架中Message模型需要更强的表现力和约束力。2.2 使用JSON Schema进行严格建模为了确保在框架内部、以及与外部AI服务API交互时数据格式的一致性我们通常会使用JSON Schema来正式定义Message的结构。JSON Schema是一种描述和验证JSON数据结构的强大工具。假设我们使用类似com.github.victools.jsonschema.generator.SchemaGenerator这样的Java库从你的热词中可见来从Message类自动生成Schema那么我们的类设计会直接影响生成的Schema。一个增强版的Message类可能如下所示以Java为例但原理通用public class Message { JsonProperty(required true) // 表明此字段在JSON中必须存在 private Role role; // 使用枚举而非字符串约束性更强 JsonProperty private String content; // 可以是字符串 JsonProperty private ListContentBlock contentBlocks; // 或者是一个复杂的内容块列表用于多模态 JsonProperty private String name; // 工具调用名称 JsonProperty private String toolCallId; // 关联的工具调用ID用于匹配调用和结果 // 枚举定义限制role的取值 public enum Role { system, user, assistant, tool } // 复杂内容块的例子 public static class ContentBlock { private String type; // text, image_url等 private Object value; // 对应的值 } }通过这个类定义SchemaGenerator会生成一个详细的JSON Schema。这个Schema会明确规定role字段是必需的且只能是system、user、assistant、tool中的一个。content和contentBlocks可能互斥或共存具体取决于业务逻辑。name和toolCallId在role为tool时可能是必需的。这么做的核心价值是什么早期错误捕获在数据进入处理管道前就能根据Schema进行验证避免将格式错误的数据发送给AI服务从而减少类似data incompatible with messages format或400 Bad Request的错误。清晰的接口契约无论是框架内部的模块还是对外提供的APISchema都作为一份“合同”明确规定了数据的形状降低了协作成本。文档化生成的Schema本身就可以作为技术文档供开发者查阅。2.3 实战中的Message处理心得在实际编码中有几点经验值得分享内容字段的灵活性很多AI服务API如OpenAI的content字段设计得非常灵活可以是null、字符串或数组。在你的框架内部我建议将其统一封装成一个更健壮的类型。例如提供一个MessageContent类它能处理字符串、复杂内容块列表甚至是在流式响应中尚未完整的“内容片段”。这样可以避免在业务代码中到处进行if (content instanceof String)...这样的类型判断。不可变性Immutability考虑Message对象一旦创建在大多数场景下不应该被修改。这符合函数式编程的思想能避免在多线程或异步流式处理中因共享可变状态导致的诡异问题。设计时可以考虑使用Builder模式或recordJava /dataclassPython来创建不可变对象。序列化/反序列化兼容性确保你的Message类能够与你选用的JSON库如Jackson, Gson以及目标AI服务的API格式完美兼容。有时服务商的字段名如tool_callsvstoolCalls或结构略有不同你可能需要编写自定义的序列化逻辑或适配器。3. ToolCall让AI“动手”的能力抽象ToolCall或FunctionCall是AI模型与外部世界交互的桥梁。它代表了模型在推理后决定执行某个具体操作如查询天气、执行计算、调用API的意图。3.1 ToolCall的核心结构一个ToolCall对象通常需要包含以下信息id (string): 本次调用的唯一标识符。在流式响应中尤为重要用于将后续返回的Tool消息结果与这次调用精确关联起来。type (string): 调用类型通常是function。function (object):name: 要调用的函数/工具的名称。arguments: 一个JSON格式的字符串包含了调用该函数所需的参数。例如AI可能返回这样一个ToolCall{ id: call_abc123, type: function, function: { name: get_current_weather, arguments: {\location\: \Beijing\, \unit\: \celsius\} } }3.2 在Message中嵌入ToolCallToolCall并不是独立存在的它需要被嵌入到Message中。通常当AI模型决定调用工具时它会生成一个role为assistant的Message但这个Message的content可能为空或包含一些思考文本同时会携带一个tool_calls数组字段。这就是数据建模的另一个关键点扩展Message模型以支持tool_calls。你的Message类需要增加一个字段public class Message { // ... 其他字段同上 JsonProperty private ListToolCall toolCalls; // 新增字段表示此条消息中AI发起的工具调用 }对应的当工具执行完毕后我们需要生成一个role为tool的Message来返回结果。这个Message的content字段存放工具执行的结果通常是JSON字符串name字段对应工具名toolCallId字段必须与发起调用的ToolCall的id一致这样框架才能正确地将结果关联回原来的调用上下文。3.3 Tool Schema的设计与注册AI模型如何知道它可以调用哪些工具呢这就需要我们定义并注册Tool Schema。它描述了工具的名称、描述、参数列表及其类型。这本质上也是一个JSON Schema。例如对于get_current_weather工具其Schema可能如下{ type: function, function: { name: get_current_weather, description: 获取指定城市的当前天气, parameters: { type: object, properties: { location: { type: string, description: 城市名称例如Beijing }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位 } }, required: [location] } } }在Eino框架中你需要提供一个机制让开发者可以方便地注册这些Tool Schema。通常框架会维护一个全局的“工具注册表”。当与AI模型交互时框架会将注册表中所有或部分工具的Schema作为上下文信息发送给模型从而“赋予”模型调用这些工具的能力。设计心得动态与静态工具注册静态注册在应用启动时通过代码或配置文件一次性注册所有工具。简单直接适用于工具集固定的场景。动态注册允许在运行时根据对话上下文动态添加或移除工具。这更灵活能实现更复杂的Agent能力调度但对框架的设计要求更高需要处理好Schema的更新和模型上下文的同步。4. 流式管道处理“涓涓细流”的数据流流式响应是现代LLM应用提升用户体验的关键。用户无需等待全部内容生成完毕就能看到逐字输出的结果。然而这对数据管道提出了挑战我们接收的不再是一个完整的Message对象而是一个个数据块Chunk。4.1 流式数据块的结构AI服务返回的流式响应每个数据块通常是一个SSEServer-Sent Events格式的数据行。解析后其JSON结构可能包含choices[0].delta: 这个对象包含了本次数据块带来的“增量”信息。delta.role: 可能出现在第一个块指明这条消息的角色。delta.content: 本次流出的文本内容片段。delta.tool_calls: 本次流出的工具调用信息片段。这是最复杂的部分因为一个ToolCall的id、function.name和function.arguments可能是分在不同的数据块里流出的。例如你可能会依次收到数据块1: {choices:[{delta:{role:assistant}}]} 数据块2: {choices:[{delta:{content:让我}}]} 数据块3: {choices:[{delta:{content:查一下}}]} 数据块4: {choices:[{delta:{tool_calls:[{index:0, id:call_xyz, function:{name:get_weather}}]}}]} 数据块5: {choices:[{delta:{tool_calls:[{index:0, function:{arguments:{\location\:\}}]}}]} 数据块6: {choices:[{delta:{tool_calls:[{index:0, function:{arguments:Shanghai\}}}]}}]}4.2 管道设计与状态管理一个健壮的流式管道需要实现一个增量聚合器Incremental Aggregator。它的核心任务是监听数据流从网络或事件源持续读取数据块。解析与提取从每个数据块中提取出delta中的role、content、tool_calls等信息。状态累积维护一个当前正在构建的Message对象我们称之为currentMessage。将流式到来的content片段不断追加到currentMessage.content。更复杂的是处理tool_calls需要根据index和id来判断当前是开始一个新的ToolCall还是在补充一个已有ToolCall的name或arguments。arguments本身是一个JSON字符串也可能被分片传输需要拼接完整后再解析。发出完成信号当流式响应结束收到[DONE]标记或连接关闭或者检测到一条完整的Message逻辑上已结束例如一个不含tool_calls的纯文本消息内容流完了就将聚合好的完整Message对象发布给下游处理器。管道设计模式 你可以将整个处理流程建模为一个反应式流Reactive Stream。使用如Project Reactor、RxJava或Java的FlowAPI。这样流式数据源、增量聚合器、工具调用执行器、结果组装器都可以作为独立的处理阶段Processor通过背压Backpressure机制优雅地处理数据流速不匹配的问题。[SSE流] - [字节解析器] - [JSON解析器] - [增量聚合器] - [完整的Message] | v [工具调用执行器] (如果Message包含ToolCall) | v [结果组装器] - [发送给用户/下一环节]4.3 流式管道中的错误处理与边界情况流式管道比批处理更脆弱需要仔细处理各种边界情况网络中断与重连流式连接可能意外中断。管道需要有能力在断连后尝试重连并尽可能从断点恢复或者至少能优雅地失败并通知用户。数据块乱序与丢失虽然不常见但在不可靠的网络中需要考虑。为数据块添加序列号或依赖更底层的可靠传输协议是解决方案。上下文长度管理从你的热词中看到了this models maximum context length is 1048576 tokens这样的错误。在长对话的流式处理中你需要持续计算整个对话历史包括流式产生的部分的token数。当接近阈值时管道需要触发“上下文窗口滑动”或“总结”等策略这个决策过程本身也可能需要与流式处理协调。工具调用的流式触发如上例所示AI可能在生成文本的中途决定调用工具。聚合器需要能识别出tool_calls片段开始出现并可能需要在工具调用完整解析出来之前就提前做一些准备工作如预加载工具甚至暂停文本内容的推送优先处理工具调用流程。5. 整合实践构建一个简单的Eino式对话轮次让我们把Message、ToolCall和流式管道串联起来看一个简化的交互流程假设我们正在构建一个支持流式响应和工具调用的AI Agent服务。5.1 初始请求与上下文构建用户提问“北京天气怎么样”前端或客户端构建一个Message列表对话历史[ {role: system, content: 你是一个有帮助的助手。}, {role: user, content: 北京天气怎么样} ]同时将已注册的get_current_weather工具的Schema也准备好。将这个列表和工具Schema一起发送给AI服务接口并请求流式响应。5.2 流式接收与聚合处理AI服务开始返回流式数据块。我们的聚合器开始工作收到第一个块delta.role为assistant创建currentMessagerole设为assistant。陆续收到多个包含delta.content的块内容可能是“正在”、“为”、“您”、“查询”…… 聚合器将其拼接到currentMessage.content。突然收到一个块其delta.tool_calls包含了一个新的ToolCall的id和function.name。聚合器在currentMessage.toolCalls列表中初始化一个新的ToolCall对象填入已收到的id和name。后续的块陆续补充这个ToolCall的function.arguments最终拼接成完整的{location: Beijing, unit: celsius}。流结束。此时聚合器产出一个完整的Message对象{ role: assistant, content: 正在为您查询, tool_calls: [ { id: call_123, type: function, function: { name: get_current_weather, arguments: {\location\: \Beijing\, \unit\: \celsius\} } } ] }5.3 工具执行与响应组装框架接收到这个包含ToolCall的完整Message后解析与路由解析tool_calls根据function.name找到注册的get_current_weather工具执行器。执行工具将arguments反序列化为参数对象调用实际的后端天气API获取结果例如{temperature: 22, condition: 晴朗}。构建Tool Response Message创建一个新的Message{ role: tool, content: {\temperature\: 22, \condition\: \晴朗\}, name: get_current_weather, tool_call_id: call_123 // 与请求的id对应 }组织新一轮请求将最初的对话历史、AI发出的工具调用消息、以及工具返回的消息三者合并为新的上下文再次发送给AI模型请求其生成面向用户的最终回答。流式返回最终答案AI模型基于工具返回的数据生成“北京当前天气晴朗气温22摄氏度”的回复并以流式形式返回。我们的管道再次进行流式聚合最终将纯文本内容流式推送给用户。在整个过程中健壮的数据模型Message,ToolCall是确保数据在不同模块间正确传递的基石而高效的流式管道是保证用户体验流畅的关键。两者结合才能构建出真正强大、易用的Eino类AI应用框架。