1. 项目概述：从混乱到有序，API响应解析的实战价值

最近在折腾各种大语言模型（LLM）的API，从OpenAI到Claude，再到国内的智谱、DeepSeek，相信很多开发者都和我有一样的感受：各家API返回的响应格式，简直是“八仙过海，各显神通”。你刚写好一套解析OpenAI choices[0].message.content 的逻辑，换到Claude API，发现它返回的是 content[0].text ；再试试Gemini，结构又变成了 candidates[0].content.parts[0].text 。这还没算上流式响应、工具调用（Function Calling）、结构化输出（JSON Mode）这些更复杂的场景。手动为每个API写一套解析逻辑，不仅重复劳动，代码也臃肿不堪，一旦某个API更新，维护起来就是一场噩梦。

正是在这种背景下，我注意到了 free-llm-api-resources 这个项目。它本身是一个收集免费或低成本LLM API资源的列表，但对我们开发者而言，其更大的价值在于它揭示了LLM生态的多样性。这种多样性直接体现在API接口的响应格式上。因此，我决定以这个项目为引子，进行一次深入的实践： 构建一个通用、健壮且可扩展的LLM API响应格式解析器 。这个解析器的目标不是调用API，而是在收到API的原始响应后，能像瑞士军刀一样，无论面对哪个厂商、哪种模式的返回数据，都能准确、一致地提取出我们需要的文本内容、工具调用信息或结构化数据。

这不仅仅是封装几个 if-else 语句。它涉及到对RESTful API设计差异的理解、对JSON数据结构的动态适配、对错误和边缘情况的统一处理，以及如何设计一个优雅的抽象层来应对未来可能出现的新API。对于任何正在或计划将LLM能力集成到自己应用中的开发者来说，这都是一个必须趟过去的“坑”。本文将分享我从设计思路到代码实现的全过程，包含大量踩坑经验和可直接复用的代码片段。

2. 核心需求与设计思路拆解

在开始编码之前，我们必须明确这个解析器要解决的核心问题，以及设计上的权衡。

2.1 核心需求解析

基于常见的LLM应用开发场景，我们的解析器需要满足以下几个核心需求：

1. 多厂商兼容性 ：必须能处理主流LLM提供商（如OpenAI、Anthropic Claude、Google Gemini、智谱GLM、DeepSeek等）的Chat Completion接口标准响应。
2. 多响应模式支持 ：
   • 普通文本完成 ：提取最终的回复文本。
   • 流式响应（Streaming） ：处理Server-Sent Events (SSE) 或分块返回的JSON数据，并能够拼接出完整的回复。
   • 工具调用（Function Calling/Tool Calls） ：准确提取模型建议调用的工具名称和参数（通常是一个JSON字符串）。
   • 结构化输出（JSON Mode） ：当要求模型返回特定JSON结构时，能解析并验证该结构。
3. 健壮性与错误处理 ：API响应可能包含错误信息（如 {"error": {"message": "..."}} ）、速率限制提示、或意料之外的结构。解析器不能轻易崩溃，而应能捕获异常，并返回格式化的错误信息供上游处理。
4. 易用性与一致性 ：对外提供简单、一致的接口。例如，一个 parse_response 函数，输入原始响应和提供商类型，输出一个标准化对象，包含 content 、 tool_calls 、 is_finished （流式场景）等字段。
5. 可扩展性 ：当有新的LLM API加入时，能够以最小的代价（如添加一个配置字典或一个新的解析类）进行扩展，而不是修改核心逻辑。

2.2 设计思路：策略模式与适配器模式

面对多样化的响应格式，最直接的想法是用一堆 if provider == “openai”: ... elif provider == “claude”: ... 。这种方法在初期快速有效，但随着支持的厂商增多，代码会变得难以维护，违反了“开闭原则”。

更优雅的设计是采用 策略模式（Strategy Pattern） 和 适配器模式（Adapter Pattern） 的结合。

• 策略模式 ：我们将每个LLM厂商的响应解析逻辑封装成一个独立的“策略”类（例如 OpenAIParser , ClaudeParser , GeminiParser ）。所有这些策略类都遵循同一个接口（例如 BaseResponseParser ），定义如 parse_content() , parse_tool_calls() 等方法。
• 适配器模式 ：我们的核心解析器（ UniversalLLMParser ）并不直接包含解析逻辑。它根据输入的“提供商”标识，动态选择并实例化对应的策略类。这个核心解析器充当了客户端代码与各种不同接口之间的适配器，对外提供统一的调用方式。

这样设计的好处是：

• 高内聚低耦合 ：每个厂商的解析逻辑变化都被隔离在各自的策略类中，互不影响。
• 易于扩展 ：要支持一个新厂商，只需新建一个实现了 BaseResponseParser 的策略类，并在工厂或映射表中注册即可，无需触动其他代码。
• 便于测试 ：每个解析策略都可以独立进行单元测试。

3. 响应格式深度解析与厂商差异

要实现通用解析，必须首先深入了解“敌人”。我们选取几个最具代表性的厂商，看看它们的响应格式究竟有何不同。

3.1 OpenAI / 兼容OpenAI格式的API

这是目前事实上的标准，许多国内外的API服务都兼容此格式。

普通响应示例：

{
  “id”: “chatcmpl-123”,
  “object”: “chat.completion”,
  “created”: 1677652288,
  “model”: “gpt-4”,
  “choices”: [{
    “index”: 0,
    “message”: {
      “role”: “assistant”,
      “content”: “Hello there!”,
      “tool_calls”: [{
        “id”: “call_abc123”,
        “type”: “function”,
        “function”: {
          “name”: “get_weather”,
          “arguments”: “{\”city\”: \”Beijing\”}”
        }
      }]
    },
    “finish_reason”: “stop”
  }],
  “usage”: {“prompt_tokens”: 9, “completion_tokens”: 12, “total_tokens”: 21}
}

关键路径 ： choices[0].message.content 是文本内容。工具调用信息在 choices[0].message.tool_calls 里。

流式响应（SSE）示例： 每个数据块像这样：

data: {"id":"...","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"index":0,"finish_reason":null}]}

data: {"id":"...","object":"chat.completion.chunk","choices":[{"delta":{"content":" there"},"index":0,"finish_reason":null}]}

data: {"id":"...","object":"chat.completion.chunk","choices":[{"delta":{},"index":0,"finish_reason":"stop"}]}

data: [DONE]

解析要点 ：需要过滤掉 data: [DONE] ，解析每个JSON块，并持续拼接 choices[0].delta.content 。最后一个块的 finish_reason 不为空表示结束。

3.2 Anthropic Claude (Messages API)

Claude的API设计理念不同，它围绕“消息”和“内容块”构建。

普通响应示例：

{
  “id”: “msg_013Zva2CMHLNnXjNJJKqJ2EF”,
  “type”: “message”,
  “role”: “assistant”,
  “content”: [
    {“type”: “text”, “text”: “Hello!”},
    {“type”: “tool_use”, “id”: “toolu_01”, “name”: “get_weather”, “input”: {“city”: “Beijing”}}
  ],
  “model”: “claude-3-5-sonnet-20241022”,
  “stop_reason”: “end_turn”,
  “usage”: {“input_tokens”: 10, “output_tokens”: 15}
}

关键路径 ：文本内容在 content 数组中，寻找 type 为 “text” 的块，其 “text” 字段即回复。工具调用是 type 为 “tool_use” 的块，参数直接在 “input” 字段中，已是JSON对象，无需二次解析。这与OpenAI将参数作为JSON字符串的设计截然不同。

流式响应示例： Claude的流式响应更复杂，包含多种事件类型（ message_start , content_block_start , content_block_delta , content_block_stop , message_delta , message_stop ）。我们需要监听 content_block_delta 事件中的 delta.text 来拼接文本，并通过 message_stop 判断结束。

3.3 Google Gemini API

Gemini的API结构又有其独特之处。

普通响应示例：

{
  “candidates”: [
    {
      “content”: {
        “parts”: [{“text”: “Hello!”}],
        “role”: “model”
      },
      “finishReason”: “STOP”,
      “safetyRatings”: [...]
    }
  ],
  “usageMetadata”: {“promptTokenCount”: 5, “candidatesTokenCount”: 2}
}

关键路径 ：文本内容在 candidates[0].content.parts[0].text 。注意，工具调用（Function Calling）在Gemini中是通过 parts 中的 functionCall 对象来体现的，与内容文本并列。

3.4 国内厂商（以智谱GLM、DeepSeek为例）

国内厂商的API大多借鉴但又不完全等同于OpenAI格式，存在一些“方言”。

智谱GLM示例：

{
  “id”: “123”,
  “choices”: [{
    “index”: 0,
    “message”: {
      “role”: “assistant”,
      “content”: “你好！”
    },
    “finish_reason”: “stop”
  }],
  “usage”: {“total_tokens”: 20}
}

注意 ：智谱的响应结构非常接近OpenAI，但字段名可能全为小写，且早期版本可能没有 tool_calls 字段，需要根据具体版本文档确认。

DeepSeek示例： DeepSeek的响应格式与OpenAI高度兼容，但在流式响应和非流式响应的结构上保持完全一致，这点比较友好。其工具调用也遵循OpenAI的 tool_calls 格式。

3.5 错误响应格式

统一错误处理至关重要。尽管内容不同，但错误响应通常有规律可循：

• OpenAI风格 ： {“error”: {“message”: “You didn’t provide an API key...”, “type”: “invalid_request_error”}}
• 通用HTTP错误 ：简单的 {“message”: “Invalid API Key”} 或 {“detail”: “Rate limit exceeded”} 。
• 带错误码的格式 ： {“code”: 402, “msg”: “Insufficient balance”} （常见于预付费或额度不足）。

我们的解析器需要首先判断响应是否包含错误信息，如果是，则应提前终止解析流程，将错误信息向上抛出。

4. 通用解析器的设计与实现

理论分析完毕，现在进入实战环节。我们将用Python来实现这个通用解析器。

4.1 定义基础接口与数据结构

首先，我们定义标准化输出的数据类（使用Pydantic或dataclass）和解析器基类。

from abc import ABC, abstractmethod
from typing import Optional, List, Any, Dict
from dataclasses import dataclass
import json

@dataclass
class ParsedResponse:
    """解析后的标准化响应"""
    content: str = “”  # 拼接后的文本内容
    tool_calls: List[Dict] = None  # 工具调用列表，每个元素包含 name, arguments(id等)
    finish_reason: Optional[str] = None  # 停止原因，如 “stop”, “length”, “tool_calls”
    is_finished: bool = True  # 当前响应是否代表一个完整回复（流式场景下有用）
    raw_response: Any = None  # 原始响应对象，供调试使用
    error: Optional[Dict] = None  # 如果解析出错或API返回错误，存放错误信息

    def __post_init__(self):
        if self.tool_calls is None:
            self.tool_calls = []

class BaseResponseParser(ABC):
    """所有厂商解析策略的基类"""
    provider_name: str

    @abstractmethod
    def parse(self, raw_response: Any, is_stream: bool = False) -> ParsedResponse:
        """核心解析方法。
        Args:
            raw_response: 可能是requests.Response对象、字典、或字符串。
            is_stream: 是否为流式响应。
        Returns:
            ParsedResponse 对象。
        """
        pass

    def _parse_json(self, raw_data: Any) -> Dict:
        """辅助方法：将多种可能的输入转换为字典。"""
        if isinstance(raw_data, dict):
            return raw_data
        if isinstance(raw_data, str):
            try:
                return json.loads(raw_data)
            except json.JSONDecodeError:
                # 可能是流式响应中的非JSON行，如 “data: [DONE]”
                return {}
        # 如果是Response对象，尝试读取JSON
        if hasattr(raw_data, ‘json’):
            try:
                return raw_data.json()
            except:
                return {}
        return {}

4.2 实现具体厂商解析策略

接下来，我们实现几个核心厂商的解析器。以OpenAI和Claude为例。

OpenAI解析器实现：

class OpenAIParser(BaseResponseParser):
    provider_name = “openai”

    def parse(self, raw_response: Any, is_stream: bool = False) -> ParsedResponse:
        parsed = ParsedResponse()
        parsed.raw_response = raw_response

        # 1. 检查是否为错误响应
        error_info = self._check_error(raw_response)
        if error_info:
            parsed.error = error_info
            return parsed

        if is_stream:
            return self._parse_streaming(raw_response)
        else:
            return self._parse_standard(raw_response)

    def _parse_standard(self, raw_response: Any) -> ParsedResponse:
        data = self._parse_json(raw_response)
        parsed = ParsedResponse()

        if not data.get(“choices”):
            # 可能是其他结构的响应，如模型列表
            parsed.content = str(data)
            return parsed

        choice = data[“choices”][0]
        message = choice.get(“message”, {})
        parsed.content = message.get(“content”, “”)
        parsed.finish_reason = choice.get(“finish_reason”)

        # 解析工具调用
        tool_calls = message.get(“tool_calls”)
        if tool_calls and isinstance(tool_calls, list):
            for tc in tool_calls:
                func_info = tc.get(“function”, {})
                parsed.tool_calls.append({
                    “id”: tc.get(“id”),
                    “type”: tc.get(“type”),
                    “function”: {
                        “name”: func_info.get(“name”),
                        “arguments”: func_info.get(“arguments”, “{}”)
                    }
                })
        return parsed

    def _parse_streaming(self, raw_response: Any) -> ParsedResponse:
        # 简化示例：假设raw_response是行迭代器
        parsed = ParsedResponse()
        parsed.is_finished = False
        content_buffer = []
        for line in raw_response.iter_lines():
            if line:
                line_decoded = line.decode(‘utf-8’).strip()
                if line_decoded.startswith(‘data: ‘):
                    event_data = line_decoded[6:] # 去掉 ‘data: ‘
                    if event_data == ‘[DONE]‘:
                        parsed.is_finished = True
                        parsed.finish_reason = “stop”
                        break
                    try:
                        chunk = json.loads(event_data)
                        choices = chunk.get(“choices”, [])
                        if choices:
                            delta = choices[0].get(“delta”, {})
                            if “content” in delta:
                                content_buffer.append(delta[“content”])
                            # 流式场景下，finish_reason在最后一个chunk
                            if choices[0].get(“finish_reason”):
                                parsed.finish_reason = choices[0][“finish_reason”]
                                parsed.is_finished = True
                    except json.JSONDecodeError:
                        continue
        parsed.content = “”.join(content_buffer)
        return parsed

    def _check_error(self, raw_response: Any) -> Optional[Dict]:
        data = self._parse_json(raw_response)
        if “error” in data:
            return {“message”: data[“error”].get(“message”, “Unknown error”), “type”: data[“error”].get(“type”), “code”: data[“error”].get(“code”)}
        # 检查HTTP错误状态码
        if hasattr(raw_response, ‘status_code’) and raw_response.status_code >= 400:
            return {“message”: f”HTTP {raw_response.status_code}”, “details”: data}
        return None

Claude解析器实现（简化版，聚焦非流式）：

class ClaudeParser(BaseResponseParser):
    provider_name = “claude”

    def parse(self, raw_response: Any, is_stream: bool = False) -> ParsedResponse:
        if is_stream:
            # 流式解析较复杂，此处省略详细实现，逻辑是监听不同事件类型
            raise NotImplementedError(“Claude streaming parser is more complex, implemented separately.”)
        return self._parse_standard(raw_response)

    def _parse_standard(self, raw_response: Any) -> ParsedResponse:
        data = self._parse_json(raw_response)
        parsed = ParsedResponse()
        # Claude错误格式可能不同
        if data.get(“type”) == “error”:
            parsed.error = {“message”: data.get(“error”, {}).get(“message”), “type”: data.get(“error”, {}).get(“type”)}
            return parsed

        content_blocks = data.get(“content”, [])
        text_parts = []
        tool_calls_list = []

        for block in content_blocks:
            block_type = block.get(“type”)
            if block_type == “text”:
                text_parts.append(block.get(“text”, “”))
            elif block_type == “tool_use”:
                tool_calls_list.append({
                    “id”: block.get(“id”),
                    “type”: “function”, # 适配通用类型
                    “function”: {
                        “name”: block.get(“name”),
                        “arguments”: json.dumps(block.get(“input”, {})) # 将对象转为字符串，与OpenAI格式对齐
                    }
                })

        parsed.content = “”.join(text_parts)
        parsed.tool_calls = tool_calls_list
        parsed.finish_reason = data.get(“stop_reason”)
        return parsed

4.3 构建解析器工厂与统一入口

最后，我们创建一个工厂来管理这些解析器，并提供统一的调用入口。

class LLMResponseParserFactory:
    _parsers: Dict[str, BaseResponseParser] = {}

    @classmethod
    def register_parser(cls, provider: str, parser_class):
        cls._parsers[provider] = parser_class()

    @classmethod
    def get_parser(cls, provider: str) -> BaseResponseParser:
        parser = cls._parsers.get(provider)
        if not parser:
            # 默认回退到OpenAI兼容解析器，因为很多API模仿它
            from .parsers.openai import OpenAIParser
            parser = OpenAIParser()
            cls._parsers[provider] = parser
        return parser

class UniversalLLMParser:
    def __init__(self):
        self.factory = LLMResponseParserFactory
        # 注册已知解析器
        self.factory.register_parser(“openai”, OpenAIParser)
        self.factory.register_parser(“claude”, ClaudeParser)
        self.factory.register_parser(“gemini”, GeminiParser) # 需实现
        self.factory.register_parser(“zhipu”, ZhipuAIParser) # 需实现
        self.factory.register_parser(“deepseek”, DeepSeekParser) # 需实现

    def parse(self, raw_response: Any, provider: str = “openai”, is_stream: bool = False) -> ParsedResponse:
        """统一解析入口"""
        parser = self.factory.get_parser(provider)
        return parser.parse(raw_response, is_stream=is_stream)

# 全局单例，方便使用
global_parser = UniversalLLMParser()

现在，在你的应用代码中，可以这样使用：

# 假设你从requests库得到了响应
import requests
response = requests.post(api_url, headers=headers, json=payload)
result = global_parser.parse(response, provider=“zhipu”)

if result.error:
    print(f”API Error: {result.error}“)
else:
    print(f”AI回复: {result.content}“)
    if result.tool_calls:
        for tool in result.tool_calls:
            print(f”建议调用工具: {tool[‘function’][‘name’]}“)

5. 实战中的陷阱与优化技巧

在实现和测试过程中，我遇到了不少坑，也总结出一些优化技巧。

5.1 流式响应的拼接与状态管理

流式解析的难点在于状态管理。你不能简单地把每个chunk的content拼起来就完事。

• 陷阱1：忽略 finish_reason 。最后一个chunk的 finish_reason 非常重要，它可能是 “stop” （正常结束）、 “length” （达到token限制）、 “tool_calls” （模型决定调用工具）。你需要把这个信息传递出去，上游逻辑可能据此决定是否要续写或执行工具。
• 陷阱2：工具调用的流式返回 。一些API在流式模式下也会返回工具调用的信息，可能是分块的。例如，OpenAI的流式响应中， delta 里也可能包含 tool_calls 字段。解析器需要能够增量式地构建完整的 tool_calls 对象，而不是只处理最后一个块。
• 优化技巧 ：为流式解析设计一个状态机或累积器。 ParsedResponse 对象在流式场景下可以设计为可变的， parse_streaming_chunk 方法每次接收一个chunk，更新内部的 content_buffer 和 partial_tool_calls ，并更新 is_finished 和 finish_reason 状态。

5.2 错误处理的边界情况

• 陷阱：网络错误与JSON解析错误 。你的解析器输入可能是 requests.Response 对象，也可能是已经解码的字典或字符串。如果网络超时， response.json() 会抛出异常。解析器最外层的 parse 方法必须用 try...except 包裹，将任何异常转化为 ParsedResponse 中的 error 信息，而不是让程序崩溃。
• 优化技巧 ：实现一个健壮的 _safe_json_loads 方法，处理各种畸形JSON，并记录日志。对于HTTP错误，除了检查JSON body里的 error 字段，一定要检查 response.status_code 。

5.3 性能与内存考虑

• 陷阱：大流式响应的内存占用 。如果模型生成了很长的文本，在内存中拼接所有chunk可能压力很大。
• 优化技巧 ：对于超长流式响应，可以提供回调（callback）接口。解析器每解析出一个完整的句子或段落，就通过回调函数输出给上游，上游可以即时处理（如显示到UI、写入文件），然后丢弃已处理的数据。这样解析器本身可以保持很低的内存占用。

5.4 扩展新厂商的标准化流程

当需要支持一个新的LLM API时，遵循以下步骤可以事半功倍：

1. 文档研究 ：仔细阅读该API的官方文档，找到Chat Completion或类似端点的响应示例（包括成功、错误、流式）。
2. 编写测试用例 ：在添加解析逻辑前，先编写单元测试。用真实的API响应示例（可以从文档或实际调用中捕获）作为测试数据。
3. 创建解析策略类 ：继承 BaseResponseParser ，实现 parse 方法。重点关注：
   • 文本内容的提取路径。
   • 工具调用信息的提取路径和格式转换（尽量统一到内部格式）。
   • 流式响应的处理逻辑（如果支持）。
   • 错误响应的识别。
4. 注册到工厂 ：在 UniversalLLMParser 的初始化部分或通过配置动态注册新的解析器。
5. 集成测试 ：进行端到端测试，模拟真实调用，确保解析器能正确工作。

6. 更进一步的思考：抽象与协议

在完成了基础解析器之后，我们可以思考更抽象的层面。我们其实是在定义一种 “LLM响应协议” 。

• 内部协议 ：我们的 ParsedResponse 就是这个内部协议的具体体现。它定义了在我们的应用内部，一个LLM响应应该长什么样。
• 外部协议适配 ：各个厂商的解析器（ OpenAIParser , ClaudeParser 等）就是外部协议到内部协议的适配器。

这种设计使得我们的应用核心逻辑与具体的LLM供应商完全解耦。未来，如果出现一个全新的、响应格式迥异的LLM API，我们只需要为其编写一个新的“适配器”，核心的业务代码（如对话管理、工具执行、上下文处理）完全不需要修改。

更进一步，我们可以将这个解析器库打包发布，比如叫做 llm-response-adapter 。它不仅可以用于你自己的项目，还可以帮助开源社区统一处理LLM API的多样性问题。你可以定义更丰富的内部协议，支持图像内容、音频内容、引用来源等更复杂的多模态输出。

通过这次基于 free-llm-api-resources 启示的实践，我们不仅解决了一个具体的工程问题，更掌握了一种应对技术生态多样性的设计方法。在快速变化的LLM领域，这种构建抽象层和适配器的能力，远比记住某个API的具体字段路径要重要得多。