1. 协议转换到底在解决什么问题第一次接触 micro-one-api 的协议转换功能是因为一个很具体的报错unexpected status 502 bad gateway: unknown error, url: http://127.0.0.1:15721/v1/responses。当时我在本地跑一个聚合网关前端用的是 Chat 格式的请求后端某个上游只认 Responses 格式中间没有任何适配层请求直接打到 502。排查了半天才意识到问题不在网络而在协议本身对不上。micro-one-api 这个项目做的事情说白了就是当翻译官。市面上主流的对话接口大致有三套协议形态Chat以messages数组为核心rolecontent结构、Responses以input和output为核心事件流式返回工具调用结构不同、MessagesAnthropic 系风格system独立字段content可以是字符串或块数组。这三套协议在字段命名、消息组织方式、工具调用tool_calls的表达、流式事件的格式上都不一样。你手上如果有一套基于 Chat 写的客户端想让它去调一个只暴露 Responses 的服务不改代码基本没戏。协议转换要解决的就是这个断层。它让 A 协议的请求进来经过一层结构映射变成 B 协议的请求发出去再把 B 的响应反向映射回 A 的格式返回给客户端。对调用方来说它以为自己在跟原生 Chat 接口说话实际上背后跑的是 Responses。这套东西适合谁三类人最需要一是做聚合网关的开发者上游五花八门下游想统一二是写客户端工具的人比如编辑器插件里同时要接多个后端三是做本地调试的手上只有一种格式的测试脚本但想验证另一种协议的服务。如果你只是单纯调一个官方接口用不上它但只要涉及多协议并存它就是刚需。我踩过的第一个坑就是以为协议转换只是改改字段名。实际上远不止工具调用、流式分片、多模态内容块每一块都有坑。下面我把整套思路和实操拆开讲。2. 三套协议的核心差异拆解2.1 Chat 协议messages 数组是绝对核心Chat 协议最典型的结构是这样{ model: xxx, messages: [ {role: system, content: 你是助手}, {role: user, content: 你好}, {role: assistant, content: 你好有什么可以帮你} ] }它的特点是所有角色平铺在一个数组里system 也是数组中的一条。工具调用的表达是 assistant 消息里带tool_calls字段然后紧跟一条role: tool的消息回填结果。这个结构简单直接但有个硬性约束带 tool_calls 的 assistant 消息后面必须跟 tool 消息否则很多服务端会直接报错就是热词里那个an assistant message with tool_calls must be followed by tool messages res。这个约束在协议转换时特别容易踩因为转换过程中如果丢了一条 tool 消息整个请求就废了。2.2 Responses 协议input 与 output 分离Responses 协议的组织方式完全不同。请求侧用input字段它可以是字符串也可以是消息数组但角色体系更精简。响应侧用output数组里面是各种类型的 item比如message、function_call、function_call_output。流式返回时是一系列事件event每个事件有type字段比如response.output_text.delta、response.function_call_arguments.delta。它和 Chat 最大的区别在于工具调用不再是嵌在消息里的字段而是独立的 output item。这意味着转换时不能简单地把tool_calls塞进某个字段而要拆成独立的 item 再重组。另外 Responses 的input对 system 的处理也不一样通常用instructions字段单独承载而不是混在消息数组里。2.3 Messages 协议system 独立content 是块数组Messages 协议Anthropic 风格的特点是system是顶层独立字段messages里只有 user 和 assistant。content 可以是纯字符串也可以是块数组比如{ role: user, content: [ {type: text, text: 看看这张图}, {type: image, source: {...}} ] }工具调用的表达是 assistant 消息里带tool_use块user 消息里带tool_result块。它和 Chat 的差异在于工具结果不是独立角色而是 user 消息里的一个块。这个差异在双向转换时是最容易出错的地方。2.4 三套协议差异对照表维度ChatResponsesMessages消息容器messages数组input/outputmessagessystemsystem 位置数组内一条instructions字段顶层独立字段工具调用表达assistant 的tool_calls独立function_callitemassistant 的tool_use块工具结果表达role: tool消息function_call_outputitemuser 的tool_result块流式事件choices[].deltaresponse.*.delta事件content_block_delta事件多模态content 数组input 内容块content 块数组看懂这张表协议转换的难点就清楚了一半。剩下的另一半是流式处理和边界情况。3. 转换层的整体架构设计3.1 为什么选中间表示层而不是两两直转三套协议两两转换理论上有 6 个方向。如果每个方向都写一套直转逻辑代码量是 6 份而且每加一套新协议就要新增 N 个方向维护成本爆炸。我的做法是引入一个中间表示层IRIntermediate Representation所有协议先转成 IR再从 IR 转成目标协议。这样 N 套协议只需要 2N 份转换代码进和出各一份扩展性完全不一样。IR 的设计原则是信息不丢失。它要能承载三套协议的所有语义包括角色、内容块、工具调用、工具结果、多模态、流式增量。我用的 IR 大致长这样class IRMessage: role: str # system / user / assistant / tool content: list # 内容块列表 tool_calls: list # 工具调用列表 tool_call_id: str # 工具结果对应的调用 id class IRContentBlock: type: str # text / image / tool_use / tool_result text: str image_url: str tool_name: str tool_input: dict tool_output: str这个结构看起来简单但它把三套协议的差异都吸收进来了。Chat 的role: tool消息转成 IR 时变成role: tooltool_call_idMessages 的tool_result块转成 IR 时也归一到同样的结构。这样反向转换时就有统一的数据源。3.2 请求方向与响应方向的对称处理转换层要处理两个方向请求转换客户端协议 → 上游协议和响应转换上游协议 → 客户端协议。这两个方向必须对称设计否则会出现请求转过去了响应转不回来的尴尬。我的做法是把转换逻辑写成一对函数to_ir(payload, protocol)和from_ir(ir, protocol)。请求方向是from_ir(to_ir(req, client_proto), upstream_proto)响应方向是from_ir(to_ir(resp, upstream_proto), client_proto)。这样无论哪个方向逻辑都是复用的不会出现两套不一致的代码。提示对称设计的关键是 IR 必须无损。如果 IR 丢了一个字段请求方向可能没事响应方向就会缺信息。我建议在 IR 里保留一个raw字段存原始 payload方便排查转换丢失问题。3.3 流式转换的特殊处理非流式转换相对简单一次性把整个 JSON 转完就行。流式转换麻烦得多因为三套协议的流式事件格式完全不同。Chat 是data: {choices:[{delta:{content:...}}]}Responses 是event: response.output_text.deltadata: {...}Messages 是event: content_block_deltadata: {...}。我的处理方式是在 IR 层定义一套统一的流式事件比如IRDelta(typetext, text...)、IRDelta(typetool_call, ...)。上游的流式事件先转成 IR 事件再转成客户端协议的事件。这样流式和非流式共用同一套 IR只是入口和出口不同。流式转换有个必须注意的点事件顺序和边界。比如 Responses 的工具调用参数是分多个 delta 传的要累积完才能转成 Chat 的完整tool_calls。如果直接一个 delta 转一个客户端会收到一堆残缺的 JSON。我一般用一个缓冲区累积遇到结束事件再一次性输出。4. 核心字段映射的实操细节4.1 system 字段的三向映射system 的处理是三套协议差异最直观的地方。Chat 里 system 是 messages 数组的一条Responses 里是instructions字段Messages 里是顶层system字段。转换规则如下Chat → Responses把role: system的消息内容抽出来放到instructions。Chat → Messages同样抽出来放到顶层system。Responses → Chat把instructions包成{role: system, content: ...}塞进 messages 开头。Messages → Chat把顶层system包成 system 消息塞进开头。看起来简单但有个坑多条 system 消息。Chat 允许有多条 systemResponses 和 Messages 只支持一个。我的处理是把多条 system 用换行拼接成一条。这个决策要谨慎因为拼接可能改变语义但对大多数场景是可接受的。4.2 工具调用的双向转换工具调用是最容易出错的部分。以 Chat → Responses 为例Chat 的 assistant 消息{ role: assistant, content: null, tool_calls: [ { id: call_abc, type: function, function: {name: get_weather, arguments: {\city\:\北京\}} } ] }转成 Responses 的 output item{ type: function_call, call_id: call_abc, name: get_weather, arguments: {\city\:\北京\} }注意id变成了call_idfunction.name提升到了顶层name。反向转换时要把这些字段还原回去。工具结果同理Chat 的role: tool消息要转成 Responses 的function_call_outputitemtool_call_id对应call_id。注意Chat 的arguments是字符串形式的 JSONResponses 也是字符串但 Messages 的tool_use的input是对象。转换时要做字符串和对象的互转别忘了json.loads和json.dumps。4.3 多模态内容块的转换多模态内容在 Chat 里是 content 数组元素形如{type: image_url, image_url: {url: ...}}。Messages 里是{type: image, source: {type: base64, media_type: ..., data: ...}}。Responses 里是{type: input_image, image_url: ...}。三者的图片表达方式不同Chat 用 URL 对象Messages 用 source 对象支持 base64 和 URLResponses 用直接的 URL 字符串。转换时要判断来源类型做相应的字段重组。base64 和 URL 的互转是最麻烦的如果上游只接受 URL 而客户端传的是 base64就得先上传拿到 URL这一步很多转换层会漏掉。4.4 参数与采样字段的映射除了消息结构采样参数也要映射。temperature、top_p、max_tokens这些字段三套协议基本一致但命名有细微差别。比如 Chat 的max_tokens在 Responses 里叫max_output_tokensMessages 里叫max_tokens。stop在 Chat 里是数组Messages 里叫stop_sequences。这些字段如果漏转行为会不一致但不报错排查起来很隐蔽。我整理了一份常用参数映射表ChatResponsesMessagesmax_tokensmax_output_tokensmax_tokensstopstopstop_sequencestemperaturetemperaturetemperaturetop_ptop_ptop_pstreamstreamstream5. 完整实操搭一个最小可用的转换网关5.1 环境准备与依赖我用 Python 做示例依赖很少主要是 web 框架和 HTTP 客户端。选 FastAPI 是因为它处理异步流式响应很方便httpx 做上游请求支持流式读取。pip install fastapi uvicorn httpx项目结构建议这样组织micro-one-api/ ├── main.py # 入口路由 ├── ir.py # 中间表示定义 ├── converters/ │ ├── chat.py # Chat - IR │ ├── responses.py # Responses - IR │ └── messages.py # Messages - IR └── stream.py # 流式转换这样分层的好处是每套协议的转换逻辑独立加新协议只加一个文件。5.2 IR 定义与转换函数骨架先定义 IR这是整个转换层的地基from dataclasses import dataclass, field from typing import Any dataclass class IRContentBlock: type: str text: str image_url: str tool_name: str tool_input: dict field(default_factorydict) tool_output: str dataclass class IRMessage: role: str content: list field(default_factorylist) tool_calls: list field(default_factorylist) tool_call_id: str dataclass class IRRequest: model: str messages: list system: str temperature: float 1.0 max_tokens: int 0 stream: bool False raw: dict field(default_factorydict)raw字段存原始 payload排查问题时能直接对比转换前后的差异这个习惯帮我省了很多时间。5.3 Chat 到 IR 的转换实现def chat_to_ir(payload: dict) - IRRequest: ir IRRequest( modelpayload.get(model, ), temperaturepayload.get(temperature, 1.0), max_tokenspayload.get(max_tokens, 0), streampayload.get(stream, False), rawpayload, ) for msg in payload.get(messages, []): role msg.get(role) if role system: ir.system msg.get(content, ) \n continue ir_msg IRMessage(rolerole) content msg.get(content) if isinstance(content, str): ir_msg.content.append(IRContentBlock(typetext, textcontent)) elif isinstance(content, list): for block in content: if block.get(type) text: ir_msg.content.append(IRContentBlock(typetext, textblock[text])) elif block.get(type) image_url: ir_msg.content.append(IRContentBlock( typeimage, image_urlblock[image_url][url])) if msg.get(tool_calls): for tc in msg[tool_calls]: ir_msg.tool_calls.append({ id: tc[id], name: tc[function][name], arguments: tc[function][arguments], }) if role tool: ir_msg.tool_call_id msg.get(tool_call_id, ) ir.messages.append(ir_msg) return ir这段代码的关键点是system 单独抽出来累积content 字符串和数组统一成块列表tool_calls 归一成统一结构。这样 IR 就与具体协议解耦了。5.4 IR 到 Responses 的转换实现def ir_to_responses(ir: IRRequest) - dict: input_items [] for msg in ir.messages: if msg.role tool: input_items.append({ type: function_call_output, call_id: msg.tool_call_id, output: msg.content[0].text if msg.content else , }) continue for block in msg.content: if block.type text: input_items.append({ role: msg.role, content: block.text, }) elif block.type image: input_items.append({ role: msg.role, content: [{type: input_image, image_url: block.image_url}], }) for tc in msg.tool_calls: input_items.append({ type: function_call, call_id: tc[id], name: tc[name], arguments: tc[arguments], }) payload { model: ir.model, input: input_items, temperature: ir.temperature, stream: ir.stream, } if ir.system: payload[instructions] ir.system.strip() if ir.max_tokens: payload[max_output_tokens] ir.max_tokens return payload这里有个细节Responses 的input数组里普通消息和 function_call item 是混在一起的顺序很重要。工具调用必须出现在对应的工具结果之前否则上游会报错。5.5 流式响应的转换处理流式转换用一个生成器函数处理边读上游边转async def stream_responses_to_chat(resp_stream): buffer {} async for line in resp_stream.aiter_lines(): if not line.startswith(data: ): continue data line[6:] if data [DONE]: yield data: [DONE]\n\n break event json.loads(data) etype event.get(type, ) if etype response.output_text.delta: chunk { choices: [{delta: {content: event.get(delta, )}}] } yield fdata: {json.dumps(chunk)}\n\n elif etype response.function_call_arguments.delta: call_id event.get(call_id) buffer.setdefault(call_id, {name: , args: }) buffer[call_id][args] event.get(delta, ) elif etype response.function_call_arguments.done: call_id event.get(call_id) chunk { choices: [{delta: {tool_calls: [{ index: 0, id: call_id, type: function, function: { name: buffer[call_id][name], arguments: buffer[call_id][args], } }]}}] } yield fdata: {json.dumps(chunk)}\n\n工具调用的参数是分片传的必须用 buffer 累积到 done 事件才能输出完整 JSON。这个逻辑如果写错客户端会收到一堆解析失败的残缺 JSON报错信息还很难定位。5.6 路由与协议协商最后把路由串起来根据请求路径或 header 决定用哪套协议app.post(/v1/chat/completions) async def chat_endpoint(req: dict): ir chat_to_ir(req) upstream_payload ir_to_responses(ir) async with httpx.AsyncClient() as client: resp await client.post(UPSTREAM_URL, jsonupstream_payload) if req.get(stream): return StreamingResponse( stream_responses_to_chat(resp), media_typetext/event-stream ) return responses_to_chat(resp.json())协议协商我一般用路径区分/v1/chat/completions走 Chat/v1/responses走 Responses/v1/messages走 Messages。这样客户端不用改直接换路径就行。6. 常见问题与排查技巧实录6.1 502 与连接类错误的定位热词里那个unexpected status 502 bad gateway: unknown error, url: http://127.0.0.1:15721/v1/responses是典型症状。502 本身是网关错误但根因往往在转换层。我的排查顺序是先确认上游服务是否真的在监听用 curl 直接打上游地址。如果上游正常检查转换后的 payload 是否符合上游协议特别是必填字段。看上游日志很多 502 其实是上游收到非法 payload 后直接断开连接。我遇到过一次转换后的input数组里 function_call item 缺了call_id上游解析失败直接断连网关就报 502。补上字段就好了。6.2 tool_calls 后缺 tool 消息的报错an assistant message with tool_calls must be followed by tool messages res这个报错根因是转换过程中丢了工具结果消息。常见场景是客户端发的 Chat 请求里assistant 带 tool_calls但 tool 消息在转换时被当成普通消息处理或者顺序被打乱。我的处理原则是转换时严格保持消息顺序assistant 的 tool_calls 和后续的 tool 消息必须成对出现。如果发现不成对宁可报错也不要静默丢弃否则上游会给出更难懂的报错。6.3 流式响应中断与事件丢失流式转换最常见的问题是事件丢失或顺序错乱。我总结了几条经验上游的[DONE]事件一定要透传否则客户端不知道流结束。工具调用的 delta 必须累积不能直接转发。如果上游用了event:行转换时要决定是否保留Chat 客户端通常只认data:行。6.4 常见问题速查表现象可能原因排查方向502 bad gateway转换后 payload 非法检查必填字段和 item 顺序tool_calls 报错tool 消息丢失或顺序错检查消息成对性流式无输出事件类型未匹配打印上游原始事件参数不生效字段名未映射对照参数映射表图片丢失多模态块未转换检查 content 块类型6.5 独家避坑技巧几个我踩过坑才总结出来的经验保留 raw 字段IR 里存原始 payload出问题时能直接 diff比猜快十倍。转换函数写单元测试三套协议两两转换手工测根本测不完写几个典型用例的测试改代码时心里有底。流式先跑非流式调试时先把 stream 关掉确认字段映射对了再开流式能省很多时间。日志打全转换前后的 payload 都打日志但注意脱敏别把敏感内容打出来。7. 协议转换的扩展与维护思路7.1 新增一套协议要改哪些地方因为用了 IR新增协议的成本很低。只需要做三件事定义该协议的to_ir和from_ir在路由里加一个入口在流式转换里加一个事件映射。不需要动其他协议的代码这是 IR 架构最大的价值。7.2 版本兼容的处理三套协议本身也在演进字段会增删。我的做法是在转换函数里对未知字段做透传而不是直接丢弃。这样即使协议升级旧代码也能兼容大部分场景。对于已知的废弃字段做兼容映射比如同时接受新旧两种命名。7.3 性能与并发注意点转换层本身是 CPU 密集型的 JSON 操作一般不是瓶颈。真正的瓶颈在上游请求。如果做聚合网关建议对上游连接做池化流式响应注意及时释放连接。我见过因为流式响应没正确关闭导致连接泄漏的案例跑一段时间就卡死。7.4 测试策略我的测试分三层单元测试覆盖每个转换函数集成测试跑完整的请求-响应链路回归测试用真实的上游做端到端验证。三套协议两两组合至少要有 6 个方向的集成测试用例。工具调用和多模态是重点覆盖对象这两块最容易出问题。8. 我在实际项目中的几点体会协议转换这东西看起来是纯技术活实际上很考验对三套协议语义的理解深度。我最初以为只要字段名对上就行结果在工具调用和多模态上栽了好几次。后来才明白转换的本质是语义对齐不是字段搬运。Chat 的role: tool和 Messages 的tool_result块语义相同但结构完全不同转换时要理解它们各自在协议里的角色才能正确映射。另一个体会是流式转换比非流式难一个数量级。非流式你拿到完整 JSON 慢慢转流式你得处理分片、累积、边界、结束事件任何一个环节出错都会导致客户端收到残缺数据。我的建议是流式逻辑一定要单独写测试用真实的流式响应做输入验证输出的事件序列是否正确。最后分享一个小技巧调试协议转换时我会写一个回环测试把 A 协议转成 B 再转回 A对比原始和还原后的 payload。如果两者一致说明转换是无损的如果不一致差异点就是 bug 所在。这个方法帮我快速定位了好几个隐蔽的字段丢失问题。这个回环测试的思路后续还可以扩展到三套协议的两两回环作为回归测试的基线。