大模型协议转换实战:Chat、Responses与Messages互转及流式处理
2026/9/23 21:58:23 网站建设 项目流程

1. 协议转换到底在解决什么问题

做过大模型应用接入的朋友大概率都遇到过这种场景:手里攒了一堆客户端,有的走 OpenAI 的 Chat Completions 格式,有的走新的 Responses 格式,还有的对接的是 Anthropic 那套 Messages 格式。每接一个新模型或者换一个上游服务,就得把请求体、响应体、流式事件全部重写一遍。写到最后你会发现,业务逻辑没多少,全耗在字段映射上了。

micro-one-api 这个项目干的事情,就是把这三种主流协议之间的转换统一到一个中间层来做。它的核心价值在于:你只需要面向一种协议写业务代码,剩下的格式适配交给转换层。比如你的客户端只会发 Chat 格式的请求,但上游只提供 Responses 接口,中间这层就负责把 Chat 的messages数组翻译成 Responses 的input结构,再把返回的output数组翻译回choices

这件事听起来简单,实际做起来坑非常多。三种协议在消息结构、工具调用、流式事件、多模态内容块上的设计哲学完全不同。Chat 是"扁平消息列表 + role 区分",Responses 是"输入输出分离 + item 类型化",Messages 是"content block 数组 + 显式角色"。你要在这三者之间做无损转换,必须对每一层的语义都有清晰的理解。

这篇文章适合三类人看:一是正在做多模型聚合网关的开发者,二是需要对接多种上游协议的后端工程师,三是想搞清楚这三种协议差异的技术负责人。我会把转换的核心逻辑、字段映射表、流式处理、工具调用这些硬骨头一块块拆开讲,尽量做到你照着就能复现。

2. 三种协议的结构差异与转换思路

2.1 Chat Completions 的扁平结构

Chat 格式是大家最熟悉的。一个典型的请求体长这样:

{ "model": "gpt-4o", "messages": [ {"role": "system", "content": "你是一个助手"}, {"role": "user", "content": "帮我查下天气"}, {"role": "assistant", "content": null, "tool_calls": [ {"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\":\"北京\"}"}} ]}, {"role": "tool", "tool_call_id": "call_1", "content": "晴,25度"} ], "tools": [{"type": "function", "function": {"name": "get_weather", "parameters": {...}}}], "stream": true }

它的特点是所有消息都塞在一个messages数组里,用role区分类型。工具调用挂在 assistant 消息的tool_calls字段上,工具结果用role: "tool"的消息回填。这个设计很直观,但有个硬性约束:tool_calls的 assistant 消息后面必须紧跟对应的 tool 消息,否则上游会直接报错,就是热词里提到的那个an assistant message with 'tool_calls' must be followed by tool messages

2.2 Responses 的输入输出分离

Responses 格式是较新的一套设计,它把"输入"和"输出"彻底分开了。请求里用input而不是messagesinput可以是一个字符串,也可以是一个 item 数组:

{ "model": "gpt-4o", "input": [ {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "帮我查下天气"}]}, {"type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": "{\"city\":\"北京\"}"}, {"type": "function_call_output", "call_id": "call_1", "output": "晴,25度"} ], "tools": [{"type": "function", "name": "get_weather", "parameters": {...}}] }

注意几个关键差异:工具调用不再是嵌套在消息里,而是独立的function_callitem;工具结果用function_call_output,通过call_id关联;内容块用input_text/output_text显式标注方向。响应侧返回的是output数组,里面混着messagefunction_call等不同类型的 item。

2.3 Messages 的内容块数组

Messages 格式(Anthropic 风格)又是另一套逻辑。它的content永远是一个 block 数组:

{ "model": "claude-3-5-sonnet", "system": "你是一个助手", "messages": [ {"role": "user", "content": [{"type": "text", "text": "帮我查下天气"}]}, {"role": "assistant", "content": [ {"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "北京"}} ]}, {"role": "user", "content": [ {"type": "tool_result", "tool_use_id": "toolu_1", "content": "晴,25度"} ]} ], "tools": [{"name": "get_weather", "input_schema": {...}}] }

它的特点是:system 是顶层独立字段,不在 messages 里;工具调用是 assistant 消息里的tool_useblock;工具结果是 user 消息里的tool_resultblock。工具定义用input_schema而不是parameters

2.4 转换的核心策略

理解了三种结构,转换思路就清晰了。micro-one-api 采用的是"中间表示层"方案:先把源协议解析成一个统一的内部结构,再从内部结构渲染成目标协议。这样做的好处是 N 种协议只需要 2N 个转换器,而不是 N² 个。

具体到字段映射,我整理了一张核心对照表:

语义ChatResponsesMessages
系统提示messages[role=system]instructions 字段system 顶层字段
用户输入messages[role=user]input 中的 message itemmessages[role=user]
工具定义tools[].functiontools[] 扁平tools[].input_schema
工具调用assistant.tool_callsfunction_call itemcontent[type=tool_use]
工具结果role=tool 消息function_call_output itemcontent[type=tool_result]
流式增量delta 字段事件类型区分事件类型区分

这张表是转换层的骨架,所有具体实现都围绕它展开。

3. 核心字段映射与实操细节

3.1 消息角色的对齐处理

角色映射看似简单,实则暗藏玄机。Chat 有systemuserassistanttool四种角色;Responses 的 message item 只有userassistantsystemdeveloper;Messages 只有userassistant,system 被提到顶层。

转换时最容易踩的坑是system 消息的位置。Chat 允许 system 消息出现在数组任意位置(虽然实践中一般放开头),但 Messages 只认顶层system字段。所以从 Chat 转 Messages 时,需要把所有role: "system"的消息内容拼接起来,塞进顶层system字段。如果有多条 system 消息,用换行符连接是常见做法。

反过来,从 Messages 转 Chat 时,要把顶层system字段还原成一条role: "system"的消息,插到 messages 数组最前面。这里有个细节:如果system字段为空或者不存在,就不要生成空的 system 消息,否则某些上游会报参数错误。

注意:Responses 的developer角色在转换到 Chat 时,建议映射为system,因为 Chat 没有 developer 概念。映射到 Messages 时同样并入顶层 system。

3.2 工具调用的双向翻译

工具调用是转换里最复杂的部分,因为三种协议的表达方式差异最大。我以"Chat 转 Responses"为例,拆解完整流程。

第一步,提取工具定义。Chat 的tools[].function结构是嵌套的,Responses 要求扁平化:

def convert_tools_chat_to_responses(chat_tools): result = [] for t in chat_tools: if t.get("type") != "function": continue fn = t["function"] result.append({ "type": "function", "name": fn["name"], "description": fn.get("description", ""), "parameters": fn.get("parameters", {}), "strict": fn.get("strict", False) }) return result

第二步,处理消息里的工具调用。Chat 的 assistant 消息里tool_calls是一个数组,每个元素有idtypefunction.namefunction.arguments。转换到 Responses 时,要拆成独立的function_callitem:

def convert_assistant_tool_calls(msg): items = [] for tc in msg.get("tool_calls", []): items.append({ "type": "function_call", "call_id": tc["id"], "name": tc["function"]["name"], "arguments": tc["function"]["arguments"] }) return items

第三步,处理工具结果。Chat 的role: "tool"消息要转成function_call_outputitem,tool_call_id对应call_idcontent对应output

反向转换(Responses 转 Chat)时,逻辑要倒过来:把function_callitem 收集起来,合并到前一条 assistant 消息的tool_calls字段;把function_call_outputitem 转成role: "tool"的消息。这里有个关键点:Responses 的 output 数组里,function_call 和 message 可能交替出现,转换时要保证 assistant 消息和 tool 消息的配对顺序正确,否则就会触发前面提到的那个报错。

3.3 多模态内容块的转换

多模态是另一个重灾区。Chat 的 content 可以是字符串,也可以是数组,数组元素类型有textimage_urlinput_audio等。Responses 用input_textinput_imageinput_audio。Messages 用textimagedocument

图片的转换尤其要注意。Chat 的image_url是一个对象{"url": "..."},支持 base64 和 http 链接。Messages 的imagesource字段,区分base64url两种类型:

def convert_image_chat_to_messages(block): url = block["image_url"]["url"] if url.startswith("data:"): # 解析 data URI header, data = url.split(",", 1) media_type = header.split(";")[0].replace("data:", "") return { "type": "image", "source": {"type": "base64", "media_type": media_type, "data": data} } return { "type": "image", "source": {"type": "url", "url": url} }

Responses 的input_image又不一样,它用image_url字段直接放字符串,同时有detail参数控制清晰度。转换时要把这些差异抹平。

实操心得:多模态转换最容易出问题的地方是 media_type 的推断。有些客户端传的 data URI 不带 media_type,这时候要根据文件头魔数判断,或者默认用image/png。我踩过的坑是直接把不带类型的 base64 传给上游,结果被拒。

3.4 参数名的批量映射

除了结构差异,参数命名也有不少坑。我整理了一份常见参数映射:

Chat 参数Responses 参数Messages 参数
max_tokensmax_output_tokensmax_tokens
temperaturetemperaturetemperature
top_ptop_ptop_p
stop(不支持)stop_sequences
n(不支持)(不支持)
streamstreamstream
tool_choicetool_choicetool_choice

注意max_tokens在 Responses 里叫max_output_tokens,这个改名很容易漏。stop在 Messages 里叫stop_sequences,而且类型是数组。n参数在 Responses 和 Messages 里都不支持,转换时要么忽略,要么报错提示。

4. 流式响应的转换实现

4.1 三种协议的流式事件模型

流式处理是转换层最难啃的部分,因为三种协议的事件模型完全不同。

Chat 的流式是 SSE,每个 chunk 长这样:

data: {"choices":[{"delta":{"content":"你"},"index":0}]} data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":""}}]}}]} data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\""}}]}}]} data: [DONE]

它的特点是所有增量都塞在choices[].delta里,工具调用的参数是分片拼接的,靠index关联。

Responses 的流式事件类型丰富得多,有response.createdresponse.output_item.addedresponse.content_part.addedresponse.output_text.deltaresponse.function_call_arguments.deltaresponse.completed等等。每个事件有明确的语义,用type字段区分。

Messages 的流式事件是message_startcontent_block_startcontent_block_deltacontent_block_stopmessage_deltamessage_stop。文本增量是text_delta,工具调用增量是input_json_delta

4.2 流式转换的状态机设计

要把 Responses 的流式转成 Chat 的流式,不能简单地一对一映射,因为事件粒度不同。我的做法是维护一个状态机,记录当前正在处理的 item 类型和索引。

class StreamConverter: def __init__(self): self.current_item_index = -1 self.current_item_type = None self.tool_call_index = -1 self.buffer = "" def convert(self, event): etype = event.get("type") if etype == "response.output_item.added": item = event["item"] if item["type"] == "function_call": self.tool_call_index += 1 self.current_item_type = "function_call" return self._make_tool_call_start(item) elif item["type"] == "message": self.current_item_type = "message" return None elif etype == "response.output_text.delta": return self._make_content_delta(event["delta"]) elif etype == "response.function_call_arguments.delta": return self._make_tool_args_delta(event["delta"]) elif etype == "response.completed": return self._make_finish_chunk(event) return None

这个状态机的核心是:Responses 的 item 生命周期事件要映射成 Chat 的 delta 累积output_item.added对应工具调用的开始(发一个带 id 和 name 的 delta),function_call_arguments.delta对应参数分片,response.completed对应finish_reason

4.3 工具调用参数的分片拼接

工具调用的参数是 JSON 字符串,流式传输时会被切成多个片段。Chat 的客户端期望收到的是分片的arguments,每个 chunk 带index。Responses 的function_call_arguments.delta事件里,delta字段就是参数片段。

转换时要保证index的连续性。我的做法是用一个计数器,每遇到一个新的function_callitem 就递增。同时要注意,Chat 的第一个工具调用 delta 必须带idfunction.name,后续的 delta 只带arguments。这个规则如果搞错,客户端会解析失败。

def _make_tool_call_start(self, item): return { "choices": [{ "delta": { "tool_calls": [{ "index": self.tool_call_index, "id": item["call_id"], "type": "function", "function": {"name": item["name"], "arguments": ""} }] }, "index": 0 }] } def _make_tool_args_delta(self, delta): return { "choices": [{ "delta": { "tool_calls": [{ "index": self.tool_call_index, "function": {"arguments": delta} }] }, "index": 0 }] }

4.4 流式转换的边界情况

实际跑起来,边界情况比正常流程还多。我列几个高频的:

第一个是空 delta。有些上游会发delta为空字符串的事件,转换时要过滤掉,否则客户端会收到一堆无意义的 chunk。

第二个是finish_reason 的时机。Chat 的finish_reason出现在最后一个 chunk,值为stoptool_callslength等。Responses 的response.completed事件里,statuscompleted,但具体原因要看incomplete_details。转换时要根据是否有工具调用来决定finish_reasontool_calls还是stop

第三个是错误事件的传递。Responses 有response.failederror事件,Messages 有error事件。转换到 Chat 时,Chat 没有标准的错误事件格式,通常的做法是发一个带error字段的 chunk,或者直接中断流并返回 HTTP 错误。我倾向于后者,因为客户端对错误 chunk 的处理逻辑不统一。

注意:流式转换一定要做超时和断连处理。上游如果卡住不发事件,转换层要主动发心跳或者超时中断,否则客户端会一直挂着。

5. 常见报错与排查实录

5.1 502 Bad Gateway 的定位思路

热词里提到的unexpected status 502 bad gateway: unknown error, url: http://127.0.0.1:15721/v1/responses,这个报错我遇到过好几次。502 本身是网关错误,但根因往往在转换层。

排查顺序是这样的:先看转换层有没有把请求体拼错,导致上游直接拒绝。最常见的拼错是input字段类型不对——Responses 的input可以是字符串或数组,如果你传了个对象,上游会返回 400 而不是 502,但如果转换层在拼接时抛了异常,网关就会返回 502。

第二步看上游服务的健康状态。127.0.0.1:15721这种本地地址,502 通常意味着上游进程挂了或者端口没监听。用curl直接打一下上游接口,确认它是否正常。

第三步看转换层的日志。micro-one-api 这类项目一般会记录原始请求和转换后的请求,对比一下就能发现字段差异。我建议在开发阶段把这两个请求都打到日志里,上线后再关掉。

5.2 工具调用配对错误的修复

an assistant message with 'tool_calls' must be followed by tool messages这个报错,根因是消息序列不合法。Chat 协议要求每个带tool_calls的 assistant 消息,后面必须紧跟数量匹配的 tool 消息。

转换时出这个错,通常是两种情况:一是从 Responses 转 Chat 时,function_callitem 和function_call_outputitem 的顺序被打乱了;二是工具结果丢失了,比如function_call_outputcall_idfunction_callcall_id对不上。

修复方法是加一个校验层,在转换完成后遍历 messages 数组,检查每个 assistant 的tool_calls是否都有对应的 tool 消息:

def validate_tool_pairing(messages): pending_calls = set() for msg in messages: if msg["role"] == "assistant" and msg.get("tool_calls"): for tc in msg["tool_calls"]: pending_calls.add(tc["id"]) elif msg["role"] == "tool": pending_calls.discard(msg["tool_call_id"]) if pending_calls: raise ValueError(f"未配对的工具调用: {pending_calls}")

这个校验放在转换之后、发送之前,能提前拦截大部分配对错误。

5.3 常见问题速查表

报错信息可能原因排查方向
502 Bad Gateway上游进程挂了 / 转换层异常先 curl 上游,再看转换日志
tool_calls 未配对消息顺序错乱 / call_id 不匹配加配对校验,检查转换顺序
400 invalid inputinput 字段类型错误确认是字符串还是数组
流式无响应上游卡住 / 事件类型未识别加超时,打印原始事件
工具参数解析失败arguments 分片拼接错误检查 index 连续性
max_tokens 超限参数名映射错误确认 max_output_tokens

5.4 独家避坑技巧

分享几个我从实际项目里总结的技巧。

第一个是保留原始请求。转换层收到请求后,先把原始 body 存一份,转换后再存一份,出问题时对比这两份能快速定位。我一般用 debug 日志级别控制,生产环境关掉。

第二个是给转换函数写单元测试。三种协议两两转换有 6 个方向,每个方向至少覆盖:纯文本、带工具调用、带多模态、流式。我写了一套 fixture,把典型请求存成 JSON 文件,测试时直接加载对比输出。

第三个是流式转换加缓冲。有些上游的事件粒度太细,一个字符一个事件,直接转发会给客户端造成压力。我加了一个 50ms 的缓冲窗口,把窗口内的事件合并后再发,实测能减少 60% 的 chunk 数量。

第四个是参数白名单。转换时不要无脑透传所有参数,而是维护一个白名单,只转发目标协议支持的字段。这样能避免因为多了个不认识的字段导致上游报错。

6. 部署与性能优化建议

6.1 转换层的部署形态

micro-one-api 这类转换层,部署形态一般有两种:一种是作为独立网关,所有请求都经过它;另一种是作为 SDK 嵌入到业务代码里。

独立网关的好处是统一管理、方便升级,缺点是增加了一跳网络开销。嵌入 SDK 的好处是性能好,缺点是每个服务都要升级依赖。我的建议是:如果上游协议经常变,用独立网关;如果协议稳定,用 SDK。

部署时要注意端口规划。热词里的127.0.0.1:15721是本地回环地址,说明转换层和上游在同一台机器上。这种部署方式延迟最低,但要注意端口冲突。我一般把转换层放在 15721,上游放在 15722,避免和常用端口撞车。

6.2 性能优化的几个抓手

转换层的性能瓶颈主要在 JSON 序列化和流式处理上。优化手段有这么几个:

第一,用流式 JSON 解析器。对于大请求体,一次性json.loads会占用大量内存,用ijson这类流式解析器能降低内存峰值。

第二,复用 HTTP 连接。转换层到上游的请求,用连接池复用 TCP 连接,能显著降低延迟。Python 里用httpx.AsyncClient或者aiohttp.ClientSession都支持连接池。

第三,流式转换避免字符串拼接。工具调用的参数分片,不要用+=拼接字符串,而是用列表收集后join,或者直接透传分片不拼接。

第四,加缓存。工具定义的转换结果可以缓存,因为同一个请求的工具定义通常不变。用functools.lru_cache或者 Redis 都行。

6.3 监控与告警

转换层上线后,必须加监控。我关注的指标有这几个:请求成功率、转换耗时 P99、上游错误率、流式中断率。

转换耗时特别重要,因为它直接叠加在用户等待时间上。如果转换耗时超过 50ms,就要排查是不是 JSON 处理太重了。

告警方面,我设置了两个阈值:上游错误率超过 5% 告警,流式中断率超过 1% 告警。这两个指标能提前发现上游故障和转换 bug。

实操心得:监控日志里一定要记录请求的协议类型和转换方向,比如chat->responses。出问题时能快速定位是哪个方向的转换出了问题。

7. 写在最后的一点个人体会

这套协议转换我前前后后迭代了三个版本。第一版是硬编码的 if-else,每加一种协议就加一堆分支,维护起来想死。第二版抽了中间表示层,代码清爽了很多,但流式处理还是各写各的。第三版把流式也统一到状态机模型,才算真正稳定下来。

如果让我给正在做类似事情的朋友一句建议,那就是:先把三种协议的字段映射表画清楚,再动手写代码。我第一版就是急着写,结果字段漏了一堆,测试的时候一个个补,反而更慢。

另外,工具调用和流式这两块,一定要单独写测试。这两块的边界情况太多了,靠手动测根本覆盖不全。我现在的做法是,每支持一种新的上游,就先跑一遍协议转换的测试集,通过了再接业务。

这个转换层后续还能扩展的方向,比如支持音频、视频等多模态,或者支持批量请求的转换。但核心思路不变:中间表示层 + 状态机流式处理,这套架构能扛住大部分协议差异。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询