数据截至 (上游 commit 3f15dc32871c)
传输与流式:统一的 HTTP 编排和增量拼装
30 秒导读: 02 章把你的参数翻译成了某家供应商能懂的 URL、headers、body。这一章讲之后发生的事:谁负责真正发出这个 HTTP 请求、连接怎么复用、错了怎么重试,以及流式响应的字节流如何被拼装成你
for循环里拿到的一个个 chunk。
1. 这一章讲什么(零基础也能懂)
一句话定义: LiteLLM 的传输层是一个通用的 HTTP 编排器,它自己不懂任何一家供应商,只负责按固定顺序去问 config「URL 是什么、header 是什么、body 长什么样」,然后发请求、收响应、拼流。
它要解决的问题
假设 LiteLLM 支持 100 家供应商。如果每家都自己写一遍「建 httpx client → 发 POST → 判断状态码 → 抛异常 → 解析 SSE」,那就是 100 份几乎一样的样板代码,而且每份的超时行为、错误类型、连接池策略都会微妙地不一样。
LiteLLM 的选择是:这些样板只写一份,写在 BaseLLMHTTPHandler(litellm/llms/custom_httpx/llm_http_handler.py:281)里;每家供应商的差异全部收敛成 02 章讲的 BaseConfig 方法。
用起来什么样
对使用者而言,传输层是完全隐形的。你只写这两种代码:
# 示意,非源码
import litellm
# 非流式:一次拿到完整回答
resp = litellm.completion(model="anthropic/claude-sonnet-4-5", messages=msgs)
print(resp.choices[0].message.content)
# 流式:一个 for 循环,增量拿
for chunk in litellm.completion(model="anthropic/claude-sonnet-4-5", messages=msgs, stream=True):
print(chunk.choices[0].delta.content or "", end="")
重点看第二段:不管底层是 Anthropic 的 event: content_block_delta、还是 Bedrock 的二进制 event stream、还是根本不支持流式的模型,你拿到的都是同一个形状的 ModelResponseStream(OpenAI 的 chunk 格式)。这一章讲的就是"怎么做到的"。
一句话直觉
把传输层想成快递分拣中心:上游各家用不同的包装(SSE、JSON、二进制帧)把货送来,分拣中心统一拆包、重新装进同一种标准箱,再一箱一箱传给收件人。收件人永远只见标准箱。
2. 顶层全景(它大概怎么转)
怎么读这张图: 从上往下是一次请求的时间顺序。左边是编排(决定调什么、按什么顺序),右边是连接(真正的 socket 层)。
litellm.completion(...) ← 01 章的主线入口
│
▼
┌───────────────────────────┐
│ BaseLLMHTTPHandler │ 编排层:唯一一份,不懂任何供应商
│ .completion / .async_* │
└───────┬───────────────────┘
│ 问 config 要 URL / headers / body ← 02 章的 BaseConfig
▼
┌───────────────────────────┐ ┌──────────────────────┐
│ _make_common_*_call │───────►│ AsyncHTTPHandler / │
│ 重试 + 统一错误封装 │ │ HTTPHandler (httpx) │
└───────┬───────────────────┘ └────── ────────────────┘
│
┌────┴─────┐
▼ ▼
非流式 流式
transform_ CustomStreamWrapper
response (第 7 节)
部件一句话职责
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
BaseLLMHTTPHandler | 编排:按固定顺序调 config 的钩子,决定走哪条分支 | litellm/llms/custom_httpx/llm_http_handler.py:281 |
_make_common_sync_call / _make_common_async_call | 真正发 POST,包住重试与错误转换 | 同上 :332 / :282 |
AsyncHTTPHandler / HTTPHandler | httpx 客户端封装:连接池、SSL、超时、错误脱敏 | litellm/llms/custom_httpx/http_handler.py:542 / :1127 |
get_async_httpx_client / _get_httpx_client | 按参数指纹缓存并复用 client(不是每次新建) | 同上 :1443 / :1497 |
| 供应商 iterator | 把该家的原始 chunk 解析成中间结构 | BaseModelResponseIterator(litellm/llms/base_llm/base_model_iterator.py:66) |
CustomStreamWrapper | 把中间结构统一成 OpenAI chunk,交给用户的 for 循环 | litellm/litellm_core_utils/streaming_handler.py:185 |
stream_chunk_builder | 反向:把一串 chunk 拼回一个完整 ModelResponse | litellm/main.py:8530 |
3. 编排:completion() 的固定七步
这节讲什么: 一次真实调用里,handler 到底按什么顺序调 02 章那些 config 方法。
七步顺序
同步入口是 BaseLLMHTTPHandler.completion(llm_http_handler.py:455)。前六步对所有分支(同步/异步、流式/非流式)都一样,只有最后一步分岔:
| # | 调的 config 方法 | 产出 | 源码行 |
|---|---|---|---|
| 1 | should_fake_stream | 这个模型要不要"装作"支持流式 | :488 |
| 2 | validate_environment | 认证后的 headers | :492 |
| 3 | get_complete_url | 完整的 api_base | :502 |
| 4 | transform_request | 供应商格式的 data(请求体) | :511 |
| 5 | sign_request | 签名后的 headers,以及可选的 signed_json_body(字节) | :522 |
| 6 | logging_obj.pre_call | 打日志(细节见 04 章) | :534 |
| 7 | 发请求 + transform_response | ModelResponse 或流对象 | :615、:646 |
第 5 步值得单独说一句:sign_request 返回的是 (headers, signed_json_body) 二元组。因为像 AWS SigV4 这类签名算法签的是请求体的确切字节,所以一旦签了名,后面就必须原样发那串字节,不能再 json.dumps 一遍。发请求时那行三元表达式正是这个意思:
data=(signed_json_body if signed_json_body is not None else json.dumps(data)),
(llm_http_handler.py:304,异步侧;同步侧在 :354。)
第 7 步的四岔口
怎么读这张图: 两个布尔量(acompletion 和 stream)组合出四条出口,每条出口的返回值类型不同。
第 1-6 步(完全相同)
│
┌─────────────┴─────────────┐
acompletion=True acompletion=False
│ │
┌─────┴─────┐ ┌─────┴─────┐
stream=T stream=F stream=T stream=F
│ │ │ │
▼ ▼ ▼ ▼
acompletion_ async_completion make_sync_ _make_common_
stream_ (:575) call (:615) sync_call (:646)
function │ │
(:556) ▼ ▼
│ CustomStream transform_response
└──► CustomStreamWrapper Wrapper → ModelResponse
注意异步分支返回的是协程而不是结果——completion() 本身是普通函数,acompletion=True 时它只是把协程对象交回去,由上层 await。
一个岔路口之外的岔路口
流式分支还有一层前置判断:如果 config 声明了 has_custom_stream_wrapper(litellm/llms/base_llm/chat/transformation.py:404),handler 就完全放手,直接把控制权交给 config 自己的 get_sync_custom_stream_wrapper / get_async_custom_stream_wrapper(llm_http_handler.py:603、:765)。这是给那些"流式协议实在太特殊、通用编排包不住"的供应商留的逃生口。
4. 发请求那一层:两个 _make_common_*_call
它要解决的小问题: 上面四条出口里有三条最终都要发一个 POST。如果每条各写一遍 try/except,错误封装迟早会走样——同一个 401,流式路径抛 AuthenticationError、非流式路径抛裸 httpx 异常,用户就没法统一处理。
思路: 把"发 POST + 错误转换 + 重试"抽成一个函数,所有路径都必须走它。源码注释把意图写得很直白:
Common implementation across stream + non-stream calls. Meant to ensure consistent error-handling.
(llm_http_handler.py:295,_make_common_async_call 的 docstring。)
同步版是 _make_common_sync_call(:332),两者逻辑逐行对应,只差 await。
翻译层内重试:「422 就 改请求再试一次」
这是 LiteLLM 一个不太显眼但很聪明的设计。
先分清两种重试:
| 重试发生在 | 谁做的 | 重试时改了什么 | 典型场景 |
|---|---|---|---|
| 翻译层内 | _make_common_*_call 的 for 循环 | 改请求体本身(丢掉不被接受的字段) | 供应商返回 422 "Extra inputs are not permitted" |
| 翻译层外 | Router 的冷却与回退(见 05 章) | 不改请求,换个 deployment 再打一次 | 429 限流、5xx |
第一种是这一节的主角。为什么值得单独做: 换个 deployment 重打一遍不会让"参数不被接受"这个问题消失,只有把那个参数删掉才会。
流程如下(从上往下,命中即停):
POST ──► 供应商返回 422 (httpx.HTTPStatusError)
│
▼
should_retry_llm_api_inside_llm_translation_on_http_error(e, params)
│
┌───── False ┴ True ──────┐
▼ ▼
_handle_error 已达 max_retry_on_unprocessable_entity_error ?
(抛供应商专属异常) │
┌── 是 ───┴── 否 ──┐
▼ ▼
_handle_error transform_request_on_
unprocessable_entity_error(e, data)
│
└──► 带新 data 回到 POST
三个钩子在 BaseConfig 里的默认实现都是"不重试":should_retry_... 返回 False(transformation.py:154)、transform_request_on_unprocessable_entity_error 原样返回(:164)、max_retry_on_unprocessable_entity_error 返回 0(:171)。因为默认是 0,循环写成 for i in range(max(max_retry..., 1))(llm_http_handler.py:299),保证至少发一次。
真实覆写者: Azure AI Foundry。它的 should_retry_llm_api_inside_llm_translation_on_http_error(litellm/llms/azure_ai/chat/transformation.py:274)靠匹配错误文本决定重不重试——"Extra inputs are not permitted"、"unknown field: parameter index is not a valid field" 等;max_retry_on_unprocessable_entity_error 返回 2(:293);transform_request_on_unprocessable_entity_error(:296)则针对不同错误文本做不同的删减,比如调 litellm.remove_index_from_tool_calls 把 tool call 里的 index 字段摘掉。
这是很典型的现实妥协:同一个 Azure AI 端点上,不同模型接受的参数集不一样,而且没有任何 API 能事先告诉你,只能撞一次墙再改。
错误统一封装:_handle_error
所有异常最终都经过 _handle_error(llm_http_handler.py:5738)。它做三件事:
- 从异常里榨出
status_code、error_text、error_headers——httpx.HTTPStatusError走e.response,别的走getattr(e, "text", str(e))。 - 没有 config 时,兜底抛通用
BaseLLMException。 - 有 config 时,交给
provider_config.get_error_class(...)——由供应商自己决定这个状态码该映射成哪个 LiteLLM 异常(:5788)。
第 3 点是「统一异常」体验的落点:同一个 401,不管来自哪家,用户 except litellm.AuthenticationError 都能接住。
5. 连接层:httpx client 为什么要缓存
它要解决的小问题: httpx.AsyncClient 内部持有一个连接池。如果每次请求都新建一个 client,每次都要重新做 TCP 握手 + TLS 握手,并且旧 client 的连接不会被复用,高并发下会把延迟和文件描述符都吃掉。
做法: 不直接 new,而是走两个工厂函数,它们按参数指纹查一个进程内缓存:
get_async_httpx_client(llm_provider, params, shared_session)(http_handler.py:1443)_get_httpx_client(params)(:1497)
指纹的拼法很朴素——把 params 的 key/value 串起来,再拼上 provider 名:
_cache_key_name = "async_httpx_client" + _params_key_name + llm_provider
(http_handler.py:1462;同步版在 :1512,没有 provider 后缀。)
命中就直接返回(:1473-1475),没命中才建新的并写回缓存,TTL 来自 _DEFAULT_TTL_FOR_HTTPX_CLIENTS = 3600(litellm/constants.py:204,注释写着"1 hour, re-use the same httpx client for 1 hour")。
实践里传进去的 params 通常只有一项 ssl_verify(见 llm_http_handler.py:409、:641),所以指纹粒度基本是「provider × SSL 配置」。超时不进指纹,因为它是每次 post(timeout=...) 单独传的。
传输层默认不是 httpx 自己
AsyncHTTPHandler.create_client(http_handler.py:581)建 httpx.AsyncClient 时会塞一个自定义 transport。_create_async_transport(:945)的默认选择是 aiohttp,_should_use_aiohttp_transport(:980)只有在 litellm.disable_aiohttp_transport = True 或环境变量 DISABLE_AIOHTTP_TRANSPORT 为真时才退回 httpx 原生 transport。源码给的理由是吞吐和延迟:
By default, we use AiohttpTransport since it offers much higher throughput and lower latency than httpx.
(http_handler.py:959。)
另外 litellm.force_ipv4 会让它改用 AsyncHTTPTransport(local_address="0.0.0.0"),注释说是为了绕开某些环境下 IPv6 引发的 ConnectionError(:1021-1023、:1118-1122)。
两个容易忽略的细节
连接断了会自动重来一次。 AsyncHTTPHandler.post 捕获 httpx.RemoteProtocolError 和 httpx.ConnectError 后,会新建一个一次性 client 走 single_connection_post_request 重发,并在 finally 里关掉它(http_handler.py:690-705)。这是为了应对连接池里那条连接已经被对端悄悄关掉的情况。
错误信息会脱敏。 httpx.HTTPStatusError 不是直接往上抛,而是先过 _raise_masked_async_error(:470)/ _raise_masked_sync_error(:450),用 mask_sensitive_info(:393)把响应文本里的敏感串遮掉,再包成 MaskedHTTPStatusError(:490)。目的是防止 API key 之类的东西随错误信息进日志。
6. 流式发起:从 HTTP 响应到供应商 iterator
这节讲什么: stream=True 时,发请求的方式和非流式差在哪,以及第一个"能产出 chunk 的东西"是怎么被造出来的。
同步侧是 make_sync_call(llm_http_handler.py:671),异步侧是 make_async_call_stream_helper(:803)。两者结构一样,做三件事:
- 拿 client,调
_make_common_*_call但带stream=True——这一层最终传到httpx的client.send(req, stream=stream)(http_handler.py:687),意思是不要把 body 全读进内存,保持连接开着。 - 把 httpx 响应的行迭代器交给 config:
response.iter_lines()(同步,:731)或response.aiter_lines()(异步,:867),调provider_config.get_model_response_iterator(...)换回一个供应商专属的迭代器。 - 打一条
post_call日志,标记"first stream response received"(:740、:875)。
异步侧多做了一步小而关键的事:
if isinstance(completion_stream, BaseModelResponseIterator):
completion_stream.http_response = response
(llm_http_handler.py:869-870。)为什么要回挂?因为 aiter_lines() 是个裸生成器,不持有 response 对象的引用,一旦用户提前 break 掉流,没人能去关那个 HTTP 响应。回挂之后 BaseModelResponseIterator.aclose()(base_model_iterator.py:73)才能真正 await self.http_response.aclose()——源码注释点明了实际收益:让 vLLM 这类后端中止继续生成,而不是白烧算力。
供应商 iterator:新旧两条路
这是 LiteLLM 演进过程留下的一道明显的地层。
| 新路子 | 旧路子 | |
|---|---|---|
| 长什么样 | 继承 BaseModelResponseIterator,只实现一个 chunk_parser(chunk: dict) | 在 CustomStreamWrapper 里写一个 handle_<provider>_chunk 方法 |
| 代码在哪 | 各供应商目录下,如 litellm/llms/triton/completion/transformation.py:306 | 全挤在 litellm/litellm_core_utils/streaming_handler.py:490-802 |
| 怎么被选中 | config 的 get_model_response_iterator 返回它 | _dispatch_provider_chunk 里一长串 elif custom_llm_provider == "..." |
| 拆行/去 SSE 前缀 | 基类统一做(base_model_iterator.py:111-133) | 每个 handler 自己 做 |
基类把最烦人的公共活儿包了:__next__ / __anext__ 负责 bytes 解码、跳过 SSE 里的空行、把 data: 前缀交给 _strip_sse_data_from_chunk 剥掉、[DONE] 直接翻译成"结束"(base_model_iterator.py:111-201)。子类因此可以短到只剩解析逻辑,比如 OpenAI 兼容端点的 OpenAIChatCompletionStreamingHandler.chunk_parser(litellm/llms/openai/chat/gpt_transformation.py:789)只是把 dict 重新装进 ModelResponseStream。
_strip_sse_data_from_chunk 本身是 CustomStreamWrapper 的静态方法(streaming_handler.py:2391),被基类反向引用——两条路共用同一段 SSE 解析,算是迁移期的一个折中。
旧路子还剩多少: streaming_handler.py 里目前有 12 个 handle_* 方法,从 handle_predibase_chunk(:490)到 handle_triton_stream(:769)。其中 handle_azure_chunk(:595)在整个仓库里 grep 只有定义、没有调用,已经是死代码。Triton 更能说明迁移状态:它同时有新的 TritonResponseIterator(triton/completion/transformation.py:306,由 :166 的 get_model_response_iterator 装配)和旧的 elif self.custom_llm_provider == "triton" 分支(streaming_handler.py:1374)。
7. 增量拼装:CustomStreamWrapper
这节讲什么: 用户 for 循环里的每一 次迭代,内部到底发生了什么。
主图:一条 SSE 字节流的一生
怎么读这张图: 主干从上往下是每个 chunk 的实时路径;右边挂出去的是"拼装器"分支,它不参与实时输出,只在流末被用来算总账。
provider 的 HTTP 响应(连接保持打开)
│ b'data: {"choices":[{"delta":{...}}]}\n'
▼
① 供应商 iterator 拆行 / 去 "data: " / json.loads
chunk_parser │
│ GenericStreamingChunk 或 ModelResponseStream
▼
② CustomStreamWrapper.chunk_creator 统一成 OpenAI chunk 形状
│ ModelResponseStream
├──────────────────────────► ③ self.chunks 累积(全量备份)
▼ │
④ 用户的 for chunk in resp: │ 流末
chunk.choices[0].delta.content ▼
⑤ stream_chunk_builder
拼回完整 ModelResponse
(成本 / 日志 / complete_response)
chunk_creator:三段流水线
chunk_creator(streaming_handler.py:1504)是每个 chunk 的必经之路,分三段:
第一段——认领。 调 _dispatch_provider_chunk(:1176),这是全文件最长的一个 if/elif 链(约 330 行)。它按优先级判断这个 chunk 属于哪一类:自定义 provider 的原生 ModelResponseStream(:1184)→ 通用结构 GenericStreamingChunk(:1208)→ 十几个供应商专属旧分支 → 兜底走 handle_openai_chat_completion_chunk(:1459)。
这一段还负责把两件事记在 wrapper 的状态上:self.received_finish_reason(收到过结束原因)和 model_response.usage(token 用量)。
第二段——修形。 主要是 tool call / function call 的兼容修补,都在 chunk_creator 内部(:1536-1600)。注释里点了两个真实的坑:Mistral 返回的 role 是 None,要补成 "assistant"(:1562-1563);Mistral 返回的 tool type 也是 None,要补成 "function"(:1565-1572)。
第三段——决定要不要吐。 调 return_processed_chunk_logic(:1018)。它有三种结局:
| 情况 | 动作 | 源码行 |
|---|---|---|
| chunk 有实质内容 | 组装成 ModelResponseStream 返回 | :1041-1088 |
| 已收到 finish_reason,且还没发过尾块 | 把 delta 清空成 Delta(content=None)、盖上 finish_reason,标记 sent_last_chunk | :1091-1130 |
| 空 chunk(纯 keepalive 之类) | 返回 None,__next__ 会 continue 掉 | :1089-1090 |
第二种情况值得注意:LiteLLM 主动构造了 OpenAI 语义里那个"带 finish_reason 的空 delta 尾块",即便上游供应商是把 finish_reason 挂在最后一个有内容的 chunk 上的。自定义 provider 分支里那段注释说得很明确:
Strip finish_reason from the content chunk so it appears only on the trailing empty-delta chunk (OpenAI spec).
(streaming_handler.py:1200-1202。)
model_response_creator:让整条流"像同一个响应"
每个出站 chunk 都由 model_response_creator(:804)造。它做的事看着琐碎,但每一条都在修一个真实 bug:
id固定。 第一次见到的response_id会被记住并盖到后续所有 chunk 上(:817-818)。created固定。 第一次的时间戳被存进self.created,后面全部复用——源码直接标了 issue 号https://github.com/BerriAI/litellm/issues/11437(:824-827)。否则同一次响应的各个 chunk 时间戳会一路递增。_hidden_params的展开顺序是承重的。 注释明写 "Spread order is load-bearing":self._base_hidden_params必须放在最后,才能盖住调用方传进来的同名键(:829-845)。
__next__ / __anext__:出口
同步出口 __next__(:1848)和异步出口 __anext__(:2056)是两份独立实现,但骨架一致:
拿下一个原始 chunk → chunk_creator → None 就 continue
│
├─ 首块:补 MCP tool 列表、记 completion_start_time
├─ 存进 self.chunks(拼装器要用)
├─ 有 usage 的 chunk:把 usage 从 chunk 里剥掉,
│ 改挂到 _hidden_params(避免被重复计一次)
└─ return
│
上游 StopIteration
│
┌─ 还没发尾块 ─► finish_reason_handler() 补一个,返回
└─ 已发过尾块 ─► stream_chunk_builder 算总账 → 真正结束
补尾块的 finish_reason_handler(:1808)逻辑很短但有个小巧思:如果最终 finish_reason 是 "stop" 而这次流里出现过 tool call,就改写成 "tool_calls"(:1817-1821)——因为很多供应商不区分这两者,但 OpenAI 客户端要靠它判断该不该去执行工具。
另外 __next__ / __anext__ 一进门就调 _check_max_streaming_duration(:276),超过 LITELLM_MAX_STREAMING_DURATION_SECONDS 就抛 litellm.Timeout——这是流式场景的"整体超时",区别于单次 HTTP 的连接超时。
8. 假流式:让不支持流的模型也能被 for 循环
它要解决的小问题: 有些模型压根不提供流式接口(Azure 的 o 系列早期版本、Bedrock 的部分 invoke 模型、Vertex 上的 Gemma)。如果直接报错,用户的代码就得为这几个模型写分支。
思路: 照常发一次非流式请求,把完整响应切成恰好一个 chunk,伪装成流。用户的 for 循环照样跑,只是只转一圈。
三个部件配合完成:
| 部件 | 职责 | 位置 |
|---|---|---|
BaseConfig.should_fake_stream | 声明"这个模型要假流",默认 False | litellm/llms/base_llm/chat/transformation.py:119 |
_add_stream_param_to_request_body | 假流时把 stream 从请求体里摘掉 | llm_http_handler.py:881 |
MockResponseIterator | 只产出一个 chunk 就 StopIteration | litellm/llms/base_llm/base_model_iterator.py:204 |
真流与假流的路径差异:
| 真流式 | 假流式 | |
|---|---|---|
请求体里的 stream | 加上 True(若 config 支持) | 移除 |
httpx 的 stream= | True(保持连接) | False(一次读完) |
是否调 transform_response | 不调 | 调,先拿到完整 ModelResponse |
| 产出 chunk 的东西 | get_model_response_iterator(...) | MockResponseIterator |
| chunk 数量 | N | 1 |
分岔点在 make_sync_call:671-744(异步同构,:803-866)。注意 stream = True; if fake_stream is True: stream = False 这两行——假流式在 HTTP 层就是一次普通请求。
MockResponseIterator 本身极简:一个 is_done 布尔,第一次调用返回 chunk 并置位,第二次直接停(base_model_iterator.py:217-231)。真正的转换在 convert_model_response_to_streaming(:19),它把 choices[i].message 改名成 choices[i].delta,再把 usage 挂上去——注释说这是为了让 Vertex Gemma 这类假流响应仍然能报出 token 数(:56-60)。
_add_stream_param_to_request_body 顺手还处理了另一件事:有些供应商(注释点名 Bedrock invoke)根本不接受请求体里的 stream 字段,靠 supports_stream_param_in_request_body(transformation.py:408,默认 True)来声明。
9. 反向操作:stream_chunk_builder 把流拼回去
它要解决的小问题: 流式响应是碎的,但很多事情必须对着"完整答案"才能做——算成本要总 token 数、写日志要完整文本、guardrail 要看全文才能判断。
入口: litellm.stream_chunk_builder(chunks, messages, logging_obj)(litellm/main.py:8530),核心逻辑在 ChunkProcessor(litellm/litellm_core_utils/streaming_chunk_builder_utils.py:176)。
谁在用它
| 调用方 | 目的 | 位置 |
|---|---|---|
CustomStreamWrapper 流末 | 算最终 usage,喂给成功日志与缓存 | streaming_handler.py:1931、:2183 |
@client 装饰器 | 用户传了 complete_response=True,直接把流吃干净再返回整体 | litellm/utils.py:1366-1370 |
| Router | 流式请求的成本与日志核算 | litellm/router.py:2284、:2829 |
| 各种 guardrail | 拿全文做内容审查 | 如 litellm/proxy/guardrails/guardrail_hooks/presidio.py:1124 |
complete_response=True 的实现朴素得可爱——就是一个 for 把 chunk 全收进 list,再交给 builder:
if "complete_response" in kwargs and kwargs["complete_response"] is True:
chunks = []
for idx, chunk in enumerate(result):
chunks.append(chunk)
return litellm.stream_chunk_builder(chunks, messages=kwargs.get("messages", None))
(litellm/utils.py:1366-1370。)
拼装本身
ChunkProcessor 先按 _hidden_params["created_at"] 排序(streaming_chunk_builder_utils.py:182-208)——因为并发场景下 chunk 到达顺序未必等于生成顺序。
然后 build_base_response(:295)搭出骨架,再按内容类型分别合并:get_combined_content(:570)、get_combined_tool_content(:390)、get_combined_thinking_content(:589)、get_combined_audio_content(:650),最后 calculate_usage(:919)算总量。
stream_chunk_builder 里还有一条快路径:先扫一遍 chunk,只要没出现 tool_calls / function_call / reasoning_content / thinking_blocks / annotations / audio / images / provider_specific_fields 中的任何一种,就认定是"纯文本流",直接 "".join(...) 收工(litellm/main.py:8565-8610)。绝大多数请求走的是这条。
一个防御性细节
CustomStreamWrapper.__next__ 把 stream_chunk_builder 包在 try 里,失败了就退回 calculate_total_usage(chunks=self.chunks) 兜底。注释解释得很具体:
stream_chunk_builder can re-raise (as APIError) on large agentic streams. ... it would escape
__next__and drop the request from SpendLogs.
(streaming_handler.py:1936-1942。)也就是说,拼装失败不能连累计费——宁可拿个粗略的 usage,也不能让这次请求从账单里消失。
10. 巧妙之处(可借鉴的技术)
编排与翻译彻底分离。 100 家供应商共用一份 BaseLLMHTTPHandler,差异全在 config。加一家新供应商不需要碰传输层一行代码(llm_http_handler.py:281)。
"发请求"只有一个入口。 三条业务路径都必须走 _make_common_*_call,所以错误封装不可能走样(:282、:332)。
翻译层内重试改的是请求而不是目标。 422 类错误换机器重打没用,只有删字段有用;LiteLLM 把这条单独做成了三个钩子(transformation.py:154、:164、:171)。
client 按指纹缓存 + TTL。 复用连接池而不是每次新建,TTL 1 小时兜住配置漂移(http_handler.py:1443、constants.py:204)。
流的提前中止能真正断开上游。 把 httpx response 回挂到 iterator 上,aclose() 才有东西可关,vLLM 这类后端会停止生成(llm_http_handler.py:869、base_model_iterator.py:73)。
假流式把"不支持流"这个能力差异藏干净了。 一次非流式请求 + 一个只转一圈的迭代器,用户代码零分支(base_model_iterator.py:204)。
usage 只算一次。 带 usage 的 chunk 会被重建一份、把 usage 从 chunk body 里剥掉改挂 _hidden_params,避免下游同时从两处读到而重复计费(streaming_handler.py:1902-1913、:2105-2113)。
拼装失败不连累计费。 见上一节的兜底逻辑(streaming_handler.py:1936-1950)。
11. 边界与局限
streaming_handler.py 是典型的历史地层。 _dispatch_provider_chunk 是一条约 330 行的 if/elif 链(:1176-1503),里面混着通用结构判断和十几个供应商硬编码分支。旧的 handle_*_chunk 方法还有 12 个(:490-802),其中 handle_azure_chunk(:595)全仓库只有定义没有调用——已经是死代码。
新旧两套并存,不是过渡完成时。 Triton 同时存在 TritonResponseIterator(triton/completion/transformation.py:306)和 elif custom_llm_provider == "triton" 分支(streaming_handler.py:1374)。哪一条实际生效取决于 config 返回的 chunk 结构是否命中 generic_chunk_has_all_required_fields(:1208);代码里没有任何注释说明迁移计划。
同步/异步是两份手写代码,不是一份加壳。 _make_common_sync_call 与 _make_common_async_call、__next__ 与 __anext__、make_sync_call 与 make_async_call_stream_helper 都是逐行对应的双胞胎。逻辑改动必须改两处,漂移风险是结构性的。
aclose() 只有异步版。 BaseModelResponseIterator.aclose(base_model_iterator.py:73)是 async def,同步流式路径没有对应的主动关闭手段。
翻译层内重试默认全关。 只有显式覆写三个钩子的 config 才有这个行为,当前仓库里主要是 Azure AI Foundry(azure_ai/chat/transformation.py:274-308),而且它的判断依赖错误文本字符串匹配——上游一改文案就会失效。
client 缓存 key 不含 shared_session。 _cache_key_name 只由 params(实际上是 ssl_verify)和 provider 名拼成(http_handler.py:1386),而 shared_session 是另一个参数(:1449)。先用某个 session 建出的 client 会被后续不传 session 的调用命中同一条缓存 (inferred)。
本章不覆盖的: 成本计算、日志回调、缓存命中——那些发生在这一层之外的 @client 装饰器里,见 04 章;跨 deployment 的重试与回退见 05 章。
12. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 通用 HTTP 编排器(唯一一份) | litellm/llms/custom_httpx/llm_http_handler.py | BaseLLMHTTPHandler |
| 同步编排入口(七步 + 四岔口) | litellm/llms/custom_httpx/llm_http_handler.py | BaseLLMHTTPHandler.completion |
| 异步非流式编排 | litellm/llms/custom_httpx/llm_http_handler.py | BaseLLMHTTPHandler.async_completion |
| 发 POST + 422 内重试(异步) | litellm/llms/custom_httpx/llm_http_handler.py | _make_common_async_call |
| 发 POST + 422 内重试(同步) | litellm/llms/custom_httpx/llm_http_handler.py | _make_common_sync_call |
| 流式发起(同步) | litellm/llms/custom_httpx/llm_http_handler.py | make_sync_call |
| 流式发起(异步) | litellm/llms/custom_httpx/llm_http_handler.py | make_async_call_stream_helper |
假流式时摘掉 stream 字段 | litellm/llms/custom_httpx/llm_http_handler.py | _add_stream_param_to_request_body |
| 异常统一封装 | litellm/llms/custom_httpx/llm_http_handler.py | BaseLLMHTTPHandler._handle_error |
| httpx 异步客户端封装 | litellm/llms/custom_httpx/http_handler.py | AsyncHTTPHandler |
| httpx 同步客户端封装 | litellm/llms/custom_httpx/http_handler.py | HTTPHandler |
| 客户端缓存工厂(异步/同步) | litellm/llms/custom_httpx/http_handler.py | get_async_httpx_client / _get_httpx_client |
| aiohttp transport 选择 | litellm/llms/custom_httpx/http_handler.py | _create_async_transport / _should_use_aiohttp_transport |
| 错误脱敏 | litellm/llms/custom_httpx/http_handler.py | mask_sensitive_info / MaskedHTTPStatusError |
| 流式统一包装器 | litellm/litellm_core_utils/streaming_handler.py | CustomStreamWrapper |
| 单 chunk 处理流水线 | litellm/litellm_core_utils/streaming_handler.py | CustomStreamWrapper.chunk_creator |
| 供应商分支大 if/elif | litellm/litellm_core_utils/streaming_handler.py | _dispatch_provider_chunk |
| 决定要不要吐这个 chunk | litellm/litellm_core_utils/streaming_handler.py | return_processed_chunk_logic |
| 造出站 chunk(固化 id/created) | litellm/litellm_core_utils/streaming_handler.py | model_response_creator |
| 补最后那个空 delta 尾块 | litellm/litellm_core_utils/streaming_handler.py | finish_reason_handler |
| 同步/异步迭代出口 | litellm/litellm_core_utils/streaming_handler.py | CustomStreamWrapper.__next__ / __anext__ |
SSE data: 前缀剥离 | litellm/litellm_core_utils/streaming_handler.py | _strip_sse_data_from_chunk |
| 遗留手写 chunk 解析(12 个) | litellm/litellm_core_utils/streaming_handler.py | handle_openai_chat_completion_chunk / handle_triton_stream / … |
| 供应商 iterator 基类(新路子) | litellm/llms/base_llm/base_model_iterator.py | BaseModelResponseIterator |
| 假流式的单 chunk 迭代器 | litellm/llms/base_llm/base_model_iterator.py | MockResponseIterator |
| 完整响应转流式 chunk | litellm/llms/base_llm/base_model_iterator.py | convert_model_response_to_streaming |
| 流拼回完整响应(入口) | litellm/main.py | stream_chunk_builder |
| 拼装核心 | litellm/litellm_core_utils/streaming_chunk_builder_utils.py | ChunkProcessor |
| 流式相关的 config 钩子 | litellm/llms/base_llm/chat/transformation.py | should_fake_stream / get_model_response_iterator / has_custom_stream_wrapper |
| 422 内重试的真实覆写者 | litellm/llms/azure_ai/chat/transformation.py | should_retry_llm_api_inside_llm_translation_on_http_error |
继续读: 前一章 02 — 翻译层:BaseConfig 契约与参数映射 讲这一章消费的那些 config 方法是怎么来的;下一章 04 — 横切层:@client 装饰器里的日志、缓存、成本与异常 讲包在整个传输层外面的那一圈。想先看全景回到 index.md。