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数组为核心,role+content结构)、Responses(以input和output为核心,事件流式返回,工具调用结构不同)、Messages(Anthropic 系风格,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 三套协议差异对照表
| 维度 | Chat | Responses | Messages |
|---|---|---|---|
| 消息容器 | messages数组 | input/output | messages+system |
| system 位置 | 数组内一条 | instructions字段 | 顶层独立字段 |
| 工具调用表达 | assistant 的tool_calls | 独立function_callitem | assistant 的tool_use块 |
| 工具结果表达 | role: "tool"消息 | function_call_outputitem | user 的tool_result块 |
| 流式事件 | choices[].delta | response.*.delta事件 | content_block_delta事件 |
| 多模态 | content 数组 | input 内容块 | content 块数组 |
看懂这张表,协议转换的难点就清楚了一半。剩下的另一半,是流式处理和边界情况。
3. 转换层的整体架构设计
3.1 为什么选中间表示层而不是两两直转
三套协议两两转换,理论上有 6 个方向。如果每个方向都写一套直转逻辑,代码量是 6 份,而且每加一套新协议,就要新增 N 个方向,维护成本爆炸。我的做法是引入一个中间表示层(IR,Intermediate 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: "tool"+tool_call_id;Messages 的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.delta+data: {...},Messages 是event: content_block_delta+data: {...}。
我的处理方式是:在 IR 层定义一套统一的流式事件,比如IRDelta(type="text", text="...")、IRDelta(type="tool_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 允许有多条 system,Responses 和 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_id,function.name提升到了顶层name。反向转换时要把这些字段还原回去。工具结果同理,Chat 的role: "tool"消息要转成 Responses 的function_call_outputitem,tool_call_id对应call_id。
注意:Chat 的
arguments是字符串形式的 JSON,Responses 也是字符串,但 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 和 URL),Responses 用直接的 URL 字符串。转换时要判断来源类型,做相应的字段重组。base64 和 URL 的互转是最麻烦的,如果上游只接受 URL 而客户端传的是 base64,就得先上传拿到 URL,这一步很多转换层会漏掉。
4.4 参数与采样字段的映射
除了消息结构,采样参数也要映射。temperature、top_p、max_tokens这些字段三套协议基本一致,但命名有细微差别。比如 Chat 的max_tokens在 Responses 里叫max_output_tokens,Messages 里叫max_tokens。stop在 Chat 里是数组,Messages 里叫stop_sequences。这些字段如果漏转,行为会不一致,但不报错,排查起来很隐蔽。
我整理了一份常用参数映射表:
| Chat | Responses | Messages |
|---|---|---|
max_tokens | max_output_tokens | max_tokens |
stop | stop | stop_sequences |
temperature | temperature | temperature |
top_p | top_p | top_p |
stream | stream | stream |
5. 完整实操:搭一个最小可用的转换网关
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_factory=dict) tool_output: str = "" @dataclass class IRMessage: role: str content: list = field(default_factory=list) tool_calls: list = field(default_factory=list) 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_factory=dict)raw字段存原始 payload,排查问题时能直接对比转换前后的差异,这个习惯帮我省了很多时间。
5.3 Chat 到 IR 的转换实现
def chat_to_ir(payload: dict) -> IRRequest: ir = IRRequest( model=payload.get("model", ""), temperature=payload.get("temperature", 1.0), max_tokens=payload.get("max_tokens", 0), stream=payload.get("stream", False), raw=payload, ) for msg in payload.get("messages", []): role = msg.get("role") if role == "system": ir.system += msg.get("content", "") + "\n" continue ir_msg = IRMessage(role=role) content = msg.get("content") if isinstance(content, str): ir_msg.content.append(IRContentBlock(type="text", text=content)) elif isinstance(content, list): for block in content: if block.get("type") == "text": ir_msg.content.append(IRContentBlock(type="text", text=block["text"])) elif block.get("type") == "image_url": ir_msg.content.append(IRContentBlock( type="image", image_url=block["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 f"data: {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 f"data: {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, json=upstream_payload) if req.get("stream"): return StreamingResponse( stream_responses_to_chat(resp), media_type="text/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 所在。这个方法帮我快速定位了好几个隐蔽的字段丢失问题。这个回环测试的思路,后续还可以扩展到三套协议的两两回环,作为回归测试的基线。