写这篇总结之前，先交代一下背景。这是我在维护一个内部AI能力接入层时的第二轮迭代，项目代号就叫“L接入层”。第一轮做的东西能用，但代码风格很“散装”：业务方直接调HTTP工具，日志靠print，数据结构各写各的，流式响应和全量返回没有统一入口。这次的目标是把这些散装改成“封装”，往工程化收敛。DeepSeek实力不错，接口文档也友好，就拿它做了全量返回和SSE流式两个典型模式的验证样本。

这篇文字想把几个最核心的点拆开讲：为什么要在请求层统一数据结构、日志封装到底在封什么、DeepSeek接入层怎么设计才便于后续扩展，以及全量返回和流式响应两种模式的实现差异与测试方法。适合正在搭AI能力网关、或者准备把多个大模型供应商收敛到同一套接口的开发者，也适合想理解SSE到底怎么对接的人。

1. 为什么这轮非做“封装”不可

先泼盆冷水：如果你的项目只是调一两次模型API，就是为了写个Demo，那直接requests.post完全够用，不需要看这篇文章。但当调用点超过五处、调用方超过两个团队、线上开始出现超时和限流之后，散装调用带来的问题会同时爆发到面。

第一轮我们遇到的问题很典型。A同学在业务A里用线程池串行调模型，B同学在业务B里直接同步调用，两边的请求结构、错误处理、日志输出完全不一样。线上出现一次高价请求超时，排查链路需要从业务日志追到HTTP层，再手工比对进出参数，半天就没了。这还没算token消耗统计：每个调用点各写各的计数器，月底汇总时数字对不上。

所以这轮的封装，不是要把底层SDK换个写法，而是为了做到三件事：

1. 调用方只需要面对一套“模型调用接口”，不用关心具体供应商的协议细节；
2. 每个请求自动带链路信息（traceId），协议进出有统一日志，出问题了能快速锁点；
3. 全量返回和流式响应成为两个显式能力，调用方按需选用，而不是各自琢磨怎么解析。

一句话：封装的收益不是写代码时的爽快感，而是排障和扩展时的安全感。

2. 消息结构与统一响应体：先定契约再写代码

接入层的第一件事，不是选模型，而是定义数据契约。因为不管是DeepSeek还是别的供应商，请求本质上都是“一组消息+若干参数”，响应本质上都是“回复文本+元信息”。把这层契约定住，后续接十个模型也只是配置不同。

2.1 消息体：最简单元不能省

大模型接口的输入是消息列表，每条消息包含角色和内容。这个模型在OpenAI兼容接口里是通用标准，DeepSeek也遵守。我设计的内部消息结构长这样：

@dataclass class Message: role: str # system / user / assistant content: str # 文本内容

看起来简单，但有个坑：content的类型在不同接口里可能不同。大多数接口支持纯字符串，但多模态场景下可能是数组（比如图片链接、文段块）。为了兼容未来形态，内部统一用字符串，同时允许一个extra字段放扩展内容，避免一改就要动所有调用方。

另外，历史对话序列必须有序。messages是一个列表，顺序即上下文顺序。业务方经常把system提示词放最后，部分模型会不认，这里要在接入层做一次校验：system放最前面，重复的system合并，空content直接报错。

2.2 响应体：给业务方一个稳定的外壳

模型接口的原始响应字段很多，但业务方真正关心的就三块：回复文本、用量统计、耗时。所以我把内部响应统一收敛为ApiResult：

@dataclass class ApiResult: code: int # 0 表示成功，非0表示业务方定义的错误码 message: str # 可读描述 data: dict | None # 内容数据，包含reply、usage等 trace_id: str # 本次调用的链路ID

这个外壳的价值在于：把“HTTP 200但业务失败”和“HTTP 500”这两种情况尽量向调用方呈现为同一个错误模型。调用方不用关心底层是400还是429还是网关超时，只需要看code和message。

实际生产里，很多接口失败是静默的。比如网络抖动导致连接被断开，HTTP层直接抛异常，业务方的第一反应是“模型返回空”，而不是“调用链路断了”。统一响应体配合日志，能帮助调用方快速区分“模型真的回了空”和“请求根本没到模型”。

2.3 请求体载荷设计：参数默认值放在封装层

请求参数方面，我单独设计了LLMConfig结构，不在业务方代码里散着传参：

@dataclass class LLMConfig: model: str = "deepseek-chat" temperature: float = 0.7 max_tokens: int = 2048 timeout: float = 60.0 stream_timeout: float = 120.0 retry_times: int = 2

temperature这种参数，放业务方手里反而容易出事。有人为了追求确定性把temperature设成0，但有些场景0反而会导致重复；有人设置成1.8，输出明显混乱。把它收进配置并在接入层做范围校验（0～2之间，超过直接警告），是工程化该做的事。

max_tokens默认值也要按场景区分。全量返回场景里模型需要一口气输出长内容，max_tokens不能太小；流式场景下通常逐块返回，早期设小一点可以让首字延迟更低。这些差异都放在封装层处理，调用方只需要告诉接入层“我要全量”还是“我要流式”。

3. 日志封装：给大模型调用装上“黑匣子”

日志是接入层里最容易做得难看的地方。第一轮我们到处print，连print(json.dumps(response, ensure_ascii=False))这种代码都有。线上日志里不但混着业务日志，还有大段大段的响应报文，肉眼根本没法扫。

这轮我用标准库logging做了一次轻量封装，思路是让日志结构化，而不是让日志话痨。

3.1 统一Logger入口，禁止裸print

第一版我直接定义了全局logger对象，但很快发现业务方还是会绕过它去print。后来干脆在接入层入口处禁止直接print，所有内部输出必须走logger，并且日志格式强制定为：

2024-11-20 10:23:45.123 | INFO | trace_id=7f8e... | event=llm.request.start | model=deepseek-chat | msg_len=128 2024-11-20 10:23:45.456 | INFO | trace_id=7f8e... | event=llm.request.done | cost_ms=333 | prompt_tokens=180 | completion_tokens=52

关键字段拆成key=value，日志采集系统可以直接索引，不需要解析人类语言。event字段特别重要，它把日志分成事件而不是一句句话。排查问题的时候，按trace_id一搜，整个调用生命周期一目了然。

3.2 请求/响应的脱敏与截断

有一件事必须做：不要完整打印API Key，也不要完整打印请求内容。接入层里我用了一个小工具函数：

def safe_truncate(text: str, limit=200) -> str: return text[:limit] + ("..." if len(text) > limit else "")

请求里的content统一截断到200字符，响应里的reply截断到300字符。API Key只保留前4位和后2位，中间用星号替代。这样既能确认每次请求大体内容，又不会把敏感信息和超长报文整段刷进日志。

踩过的一个实实在在的坑：有次线上反馈“模型乱回答”，我打开日志发现某条请求的system提示词有几千个字符，完整刷出来之后直接把日志文件撑爆了。从那之后，所有入参都走截断，没有例外。

3.3 耗时与token埋点：指标从日志里长出来

日志另一个重要功能是监测。我在封装层埋了两个计时点：

• start_time：发送HTTP请求之前；
• end_time：完整响应落盘（全量返回）或流结束（流式响应）之后。

两者相减得到cost_ms。token用量从响应体里的usage字段读取，流式场景下需要自己累计。把这些字段写进结构化日志后，后续接Prometheus只需要做一条日志采集规则，不需要改业务代码。

日志还承担着“失败链路记录”的角色。每次重试、每次超时、每次降级，都必须有独立event。这样即使调用最终失败，也能知道它是第一次就挂，还是重试之后才挂。

4. DeepSeek接入层封装：把供应商差异关进“笼子”里

DeepSeek的接口兼容OpenAI规范，这对封装来说太友好了：请求体、响应体、鉴权头，基本可以复用一套实现。但“兼容”不等于“完全一致”，有些细节还是得在封装层专门处理。

4.1 为什么选择OpenAI兼容接口作为接入标准

市面上主流大模型供应商几乎都在跟进OpenAI的接口格式，DeepSeek也不例外。这意味着我的接入层只需要维护一套协议核心，通过配置base_url、api_key、model就能切换供应商，而不是为每家写一套适配器。

我见过有些团队用各家SDK搞适配，最后项目里装了四五个SDK，版本还互相打架。纯HTTP方案用requests或httpx就够了，SDK引入的问题比解决的问题多。

4.2 Client类的核心实现

接入层的核心是一个LLMClient类，对外暴露两个方法：chat_completion和chat_stream。

class LLMClient: def __init__(self, config: LLMConfig, api_key: str, base_url: str): self._config = config self._headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } self._url = f"{base_url}/chat/completions" def chat_completion(self, messages: list[Message]) -> ApiResult: payload = self._build_payload(messages, stream=False) resp = self._post_with_retry(payload) data = resp.json() return self._parse_completion(data) def chat_stream(self, messages: list[Message]): payload = self._build_payload(messages, stream=True) resp = self._post_with_retry_for_stream(payload) for chunk in self._iter_sse(resp): yield chunk

关键点有两个。第一，_build_payload不对外开放，所有参数收敛在内部，业务方没有机会传错。第二，chat_stream是生成器，调用方拿到的不是一个“整块结果”，而是一个可遍历的事件流，这奠定了流式响应的使用方式。

4.3 重试、超时与错误分类

调用大模型接口，网络错误是常态。DeepSeek官方文档里明确有HTTP 429（频率限制）和5xx（服务端错误），这些错误应该重试；但400（参数错误）重试也没用，直接抛回业务方。

我的重试逻辑很朴素：

RETRYABLE_STATUS = {429, 500, 502, 503, 504} def _post_with_retry(self, payload): for attempt in range(self._config.retry_times + 1): try: resp = requests.post(self._url, json=payload, headers=self._headers, timeout=(5, self._config.timeout)) if resp.status_code in RETRYABLE_STATUS: wait = 2 ** attempt # 指数退避 logger.warning("retryable status", status=resp.status_code, attempt=attempt, wait=wait) time.sleep(wait) continue resp.raise_for_status() return resp except requests.Timeout: if attempt == self._config.retry_times: raise raise RuntimeError("unexpected retry exit")

timeout我拆成了connect和read两段：连接5秒，读取60秒。connect超时说明网络路径有问题，重试价值不大；read超时则可能是模型生成太慢，重试一次往往能通。

还有个细节：所有重试日志都必须带trace_id，否则一旦重试，你都不知道哪次请求成功了。

4.4 模型切换与配置管理

配置管理我用了最简单的方式：一个llm_config.json，启动时加载成LLMConfig实例。

{ "model": "deepseek-chat", "base_url": "https://api.deepseek.com", "api_key_env": "DEEPSEEK_API_KEY", "temperature": 0.7, "max_tokens": 2048 }

api_key不直接写在配置文件里，而是从环境变量读。这算是最低级的安全底线了，线上环境不能有任何明文密钥入库。团队内部推行这个约定后，至少没有再出现过密钥被提交到代码仓库的情况。

切换模型只需要改配置里的模型名和base_url。封装层不写死任何与DeepSeek强相关的逻辑，所有差异化内容都收敛在协议解析函数里。后期接新供应商，多半只需要改_parse_completion和_iter_sse。

5. 全量返回与SSE流式：两种模式从原理到测试

这是全文最核心的部分。标题里“全量返回实现测试”和“SSE流式响应实现测试”是并列的，因为它们代表了两种完全不同的响应形态。

5.1 全量返回路径：一次性拿到完整结果

全量返回的逻辑很直接：请求发出去，等模型生成完，接口一次性返回所有内容。DeepSeek的非流式响应结构大概是：

{ "id": "chatcmpl-xxx", "model": "deepseek-chat", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "你好，有什么可以帮你？" }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 12, "completion_tokens": 9, "total_tokens": 21 } }

接入层的_parse_completion需要做这几个动作：

• 从choices[0].message.content取回复文本；
• 从usage里取token统计；
• 将整个响应包装成ApiResult.data。

有一个容易踩的坑：某些模型在max_tokens被打满时，finish_reason是length而不是stop。如果调用方只看choices有没有内容而不看finish_reason，可能会把截断的回复当成完整回复。我在封装层特意把finish_reason放进data，并加上一个truncated布尔字段，方便业务方判断。

全量返回适合不需要实时展示、或者后处理逻辑是“整段拿来做”的场景。比如离线批量生成摘要、分析报告、代码审查等。它的优点是实现简单、结果确定性高；缺点是用户等待时间取决于模型生成总时长，长文本场景下体验会差。

5.2 SSE流式返回的协议与实现

SSE全称是Server-Sent Events，它就是一种基于HTTP的单向服务端推送协议。和大模型接口配合时，服务端把生成的文本切成小块，通过HTTP连接持续往下推，客户端一行一行读。这也解释了为什么DeepSeek的接口要求stream: true时返回的不是一个JSON，而是一串以data:开头的事件流。

SSE数据块的格式大致是：

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"你"},"finish_reason":null}]} data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"好"},"finish_reason":null}]} data: [DONE]

每组数据由两个换行符分隔，[DONE]表示结束。实现解析时要注意：日志采集系统或网络代理可能把多行数据合并成一行，所以严格按空行分割会更稳妥。

我实际的流式解析代码：

def _iter_sse(self, resp): for line in resp.iter_lines(decode_unicode=True): if not line: continue if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": return yield json.loads(data)

注意这里用的是requests库的iter_lines，不是一次性读全部。网络流式传输的核心就是边收边处理，不能等响应全部落地再解析，那样就失去了流的意义。

5.3 流式解析：把chunk拼成完整的回复

每次收到的delta.content只是整个回复的一小块，封装层需要把小块拼起来，并通过生成器逐块抛给调用方：

def chat_stream(self, messages: list[Message]): payload = self._build_payload(messages, stream=True) resp = self._post_with_retry_for_stream(payload) full_text = [] for chunk in self._iter_sse(resp): delta = chunk["choices"][0].get("delta", {}) piece = delta.get("content") if piece: full_text.append(piece) yield piece # 流结束后，整段文本就绪 yield self._make_stream_done("".join(full_text))

这里我特意在流结束后补了一次yield，用_make_stream_done带出完整回复和token统计。调用方的体验是：前面拿到的是一个个小片段，用于实时展示；流结束还能拿一次汇总数据，用于记账和审计。这算是一个不大不小的设计甜点，实测对调用方特别友好。

usage在流式响应里比较微妙。DeepSeek的流式接口通常在最后一个携带了usage字段的chunk里给token统计，但choices可能为空数组。解析时必须单独判断，不能假设每个chunk都有choices[0]：

if chunk.get("usage"): self._usage = chunk["usage"]

踩坑记录：我一开始只解析choices，测试时发现token统计永远是0。后来抓原始响应才看到usage藏在最后一个data块里，而那个块的choices是空的。这个坑几乎每个对接流式的人都会撞一次。

5.4 两种响应模式的测试：从curl到自动化断言

测试流式响应比全量返回要麻烦一些，因为它的输出是持续性的，拆成几个层次做：

先拿真实接口做一轮可行性验证，用curl确认服务端是否真的按SSE格式推数据：

curl -N https://api.deepseek.com/chat/completions \ -H "Authorization: Bearer sk-xxxx" \ -H "Content-Type: application/json" \ -d '{ "model": "deepseek-chat", "messages": [{"role": "user", "content": "你好，请讲个冷笑话"}], "stream": true }'

-N参数关闭curl的缓冲，让数据一到就输出。看到一列data:块，说明链路通了。这一步主要排除“服务端压根没走流式”的情况。

再写自动化测试。全量返回的断言比较简单：

def test_llm_completion(): client = LLMClient(LLMConfig(), api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=...) result = client.chat_completion([Message(role="user", content="用一句话介绍自己")]) assert result.code == 0 assert isinstance(result.data["reply"], str) assert len(result.data["reply"]) > 0 assert result.data["usage"]["total_tokens"] > 0

流式响应的自动化测试要模拟“边收边看”的视角：

def test_llm_stream(): client = LLMClient(LLMConfig(), api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=...) pieces = list(client.chat_stream([Message(role="user", content="数到十")])) # 最后一个是汇总事件，前面的是片段 assert len(pieces) >= 2 full_text = "".join(p for p in pieces if isinstance(p, str)) assert "十" in full_text

同时我本地起了一个Mock SSE服务，用预设好的SSE数据块模拟服务端，专门测试异常场景：

• 流中途断开：看封装层是否会抛异常，还是静默返回不完整结果；
• [DONE]缺失：看解析是否会死循环或挂住；
• 数据块中choices为空：看是否触发continue而不是抛IndexError。

Mock方案的好处是不花钱、不依赖网络，还能随时让服务端“表演故障”。批量CI里跑真实API既不稳定还耗token，所以我把Mock用例设为默认跑，真实API用例单独加标记按需执行。

5.5 全量与流式的选型对比

最后给一张在实际项目中做技术选型时用的对比表，封装层内部也是这样决策的：

我内部封装了两套入口，但底层共用_build_payload。调用方只需在传入参数里标记stream=True或False，剩下的交给接入层。这也是一种“配置不污染业务代码”的做法。

6. 这一轮踩过的坑和最后沉淀的经验

这轮迭代整体顺畅，但有几个坑值得专门说，因为它们在各类教程里很少被主动提及。

第一个坑是SSE的字节流编码。requests的iter_lines在默认情况下返回字节，如果接口返回UTF-8文本，直接decode_unicode=True是必要的。我第一次没加这个参数，拼接出来的中文全是乱码。看起来是小问题，却能让人排查一小时。

第二个坑是超时配置。全量返回我一开始设了全局超时30秒，结果某个稍微长一点的生成任务直接超时。后来改成连接超时和读超时分离，读超时放大到60~120秒，问题才解决。流式响应还要考虑“块间隔超时”——不是整个流多久结束，而是两个块之间最长等多久。有些模型生成停顿能超过30秒，盲目断连会误杀正常请求。

第三个坑是我前面提过的流式token统计。这个必须再强调一遍：不要在流式解析早期假设usage一定会来，也不要假设每个chunk都有choices。真正上了生产，各种奇怪的JSON结构都会出现，防御式解析不是洁癖，是保命。

第四个坑是日志打点只放在成功路径上。第一版我忘记在异常分支里写日志，结果排查问题时只能看到“没有done日志”，完全不知道是超时、限流还是参数错误。后来凡是except块，第一句必然是logger.warning(...)，把异常类型、attempt次数、trace_id全部打上。这一条看着琐碎，实际对排障效率的提升是决定性的。

如果回头看这轮迭代，我最满意的不是哪个代码模块写得多漂亮，而是接入层暴露出的接口足够窄。业务方只看到chat_completion和chat_stream，底层是DeepSeek还是别的模型、走HTTP还是SSE、重试几次，都和他们无关。这意味着以前的“每加一个模型就要动业务代码”终于变成了“只需要改配置和协议转换”。

最后分享一个小技巧：在流式响应的封装层里，我会在生成结束时附带一个事件对象，里面放完整回复、token总用量、耗时、是否有截断等信息。这样调用方既能实时渲染，又能在结束时拿到一份“审计报告”。这个设计最初只是顺手加的，后来发现它在日志归集、成本核算、效果评估上都特别好用。值得抄作业。