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而不是messages,input可以是一个字符串,也可以是一个 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数组,里面混着message、function_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² 个。
具体到字段映射,我整理了一张核心对照表:
| 语义 | Chat | Responses | Messages |
|---|---|---|---|
| 系统提示 | messages[role=system] | instructions 字段 | system 顶层字段 |
| 用户输入 | messages[role=user] | input 中的 message item | messages[role=user] |
| 工具定义 | tools[].function | tools[] 扁平 | tools[].input_schema |
| 工具调用 | assistant.tool_calls | function_call item | content[type=tool_use] |
| 工具结果 | role=tool 消息 | function_call_output item | content[type=tool_result] |
| 流式增量 | delta 字段 | 事件类型区分 | 事件类型区分 |
这张表是转换层的骨架,所有具体实现都围绕它展开。
3. 核心字段映射与实操细节
3.1 消息角色的对齐处理
角色映射看似简单,实则暗藏玄机。Chat 有system、user、assistant、tool四种角色;Responses 的 message item 只有user、assistant、system、developer;Messages 只有user和assistant,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是一个数组,每个元素有id、type、function.name、function.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_id,content对应output。
反向转换(Responses 转 Chat)时,逻辑要倒过来:把function_callitem 收集起来,合并到前一条 assistant 消息的tool_calls字段;把function_call_outputitem 转成role: "tool"的消息。这里有个关键点:Responses 的 output 数组里,function_call 和 message 可能交替出现,转换时要保证 assistant 消息和 tool 消息的配对顺序正确,否则就会触发前面提到的那个报错。
3.3 多模态内容块的转换
多模态是另一个重灾区。Chat 的 content 可以是字符串,也可以是数组,数组元素类型有text、image_url、input_audio等。Responses 用input_text、input_image、input_audio。Messages 用text、image、document。
图片的转换尤其要注意。Chat 的image_url是一个对象{"url": "..."},支持 base64 和 http 链接。Messages 的image用source字段,区分base64和url两种类型:
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_tokens | max_output_tokens | max_tokens |
| temperature | temperature | temperature |
| top_p | top_p | top_p |
| stop | (不支持) | stop_sequences |
| n | (不支持) | (不支持) |
| stream | stream | stream |
| tool_choice | tool_choice | tool_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.created、response.output_item.added、response.content_part.added、response.output_text.delta、response.function_call_arguments.delta、response.completed等等。每个事件有明确的语义,用type字段区分。
Messages 的流式事件是message_start、content_block_start、content_block_delta、content_block_stop、message_delta、message_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 必须带id和function.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,值为stop、tool_calls、length等。Responses 的response.completed事件里,status是completed,但具体原因要看incomplete_details。转换时要根据是否有工具调用来决定finish_reason是tool_calls还是stop。
第三个是错误事件的传递。Responses 有response.failed和error事件,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_output的call_id和function_call的call_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 input | input 字段类型错误 | 确认是字符串还是数组 |
| 流式无响应 | 上游卡住 / 事件类型未识别 | 加超时,打印原始事件 |
| 工具参数解析失败 | 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,每加一种协议就加一堆分支,维护起来想死。第二版抽了中间表示层,代码清爽了很多,但流式处理还是各写各的。第三版把流式也统一到状态机模型,才算真正稳定下来。
如果让我给正在做类似事情的朋友一句建议,那就是:先把三种协议的字段映射表画清楚,再动手写代码。我第一版就是急着写,结果字段漏了一堆,测试的时候一个个补,反而更慢。
另外,工具调用和流式这两块,一定要单独写测试。这两块的边界情况太多了,靠手动测根本覆盖不全。我现在的做法是,每支持一种新的上游,就先跑一遍协议转换的测试集,通过了再接业务。
这个转换层后续还能扩展的方向,比如支持音频、视频等多模态,或者支持批量请求的转换。但核心思路不变:中间表示层 + 状态机流式处理,这套架构能扛住大部分协议差异。