从0打造编码Agent 阅读 + 实验 ≈ 35 分钟 页内实验 ×1
A8 模型接入:换模型、控成本怎么办? 第 2 课

流式回来的工具参数,什么时候才能解析?

流式工具参数何时能解析
本机服务的真实流里,两个并行的 read_file 各来一帧;把它们的参数塞进同一个缓冲,拼出来的 {…}{…} 不是合法 JSON。另一个朴素做法「每来一段就 parse」在本机录制上一次也没失败——因为这家服务把整个调用攒完才发;按 Anthropic 官方文档构造的事件流上,它从第一段空串起就连着失败。现代做法:每个调用一个缓冲,收到这个调用的结束信号才解析;Codex 更干脆,忽略参数增量,只收完整的输出项,并在流还没结束时就把工具排进执行队列。
❓

本课一问:第 1 课看的是整条回复。可 agent 几乎都用流式接口:回复一段一段地到,工具参数也是分段来的。参数什么时候才能 JSON.parse、什么时候才能开始执行?

这一课先不碰:首 token 时间、吞吐这些指标的机制(LLM 课 C13 第 5 课 讲过,这里只用);出错、限流怎么重试(岔路 B1);账单(第 3 课)。

从已有出发

还是第 1 课那条真实回复:模型在一次回复里发了两个 read_file。这一次看它是怎么一帧一帧到的。A1 第 3 课里「服务端报 tool_calls 却没给调用」的那几轮,在这一课的时间轴上也会看到它们的样子。

从已有出发 · 累计数据流 点已学环节看它回答什么
每一轮循环都经过

本机服务的流(OpenAI Chat 兼容)每一帧是一行 data: {...},最后一帧是 data: [DONE];两家云端的流还多一行 event:,事件名与 data 里的 type 相同。录制时发现:mlx_lm.server 不增量发送工具参数:整个 <tool_call> 生成完才发一帧完整的 tool_calls(带 index);content / reasoning 按 token 增量发送。

先自己设计:收到就解析,或者一个缓冲装全部

两个最容易想到的写法:① 每收到一段参数就 JSON.parse 一次,成功了就用;② 回复里所有工具参数都往同一个字符串里追加,流结束再解析。

  1. ① 朴素方案
  2. ② 它在哪失败
  3. ③ 改进方案
朴素方案每来一段就 JSON.parse;或所有调用共用一个缓冲
朴素拼装:在真实帧与按文档构造的事件上各跑一遍
① 原始事件
② 参数缓冲区
③ 解析结果
🎯

试三件事:① 在「并行调用 · 真实录制」上把策略切到「每来一段就 JSON.parse」:结论行说它一次也没失败——看左栏,两个调用各只有一帧,参数是整段到的;② 切到「Anthropic · 按文档构造」,同一个策略:第一个片段是空串,JSON.parse('') 就报错,之后每一段都是半截 JSON;③ 回到「共用一个缓冲」,在 Anthropic 源上拖到最后:两个 tool_use 块的参数首尾相接,和真实录制上的失败一模一样。

改进:每个调用一个缓冲,结束才解析

Anthropic 的流式文档写明:input_json_delta 是不完整的 JSON 片段,最终的 tool_use.input 才是对象;可以累积到 content_block_stop 再解析 。OpenAI 的 function-calling 指南同样按 output_index 累积 function_call_arguments.delta,.done 事件给出完整参数。

参数「怎么分段到」本身也是一个设计选择,四家各不相同:

谁参数怎么到什么时候能解析出处
本机 mlx_lm.server(Chat 兼容)整个调用生成完,一帧给出完整参数这一帧到了就行,但只有整条回复结束才知道没有别的调用了T5 录制(项目实测)
Anthropic,标准流式攒够一个完整的键值,再切成几段 input_json_delta 发这个块的 content_block_stop官方流式文档
Anthropic,细粒度流式边生成边发,不缓冲、不校验,可能不合法同上,且解析要防御官方 fine-grained tool streaming 文档
OpenAI Responsesfunction_call_arguments.delta 逐段,.done 与 output_item.done 给完整值.done 或 output_item.done官方 function-calling 指南

Codex 用的是 Responses,可它根本不拼参数增量:参数增量事件被它列进了一张「认识但不处理」的表,它只收 output_item.done 里那个完整的调用项。

Codex:每种事件走哪一支(参数增量在忽略表里) codex@8cabf5a:codex-rs/codex-api/src/sse/responses.rs:330-495需联网 Apache-2.0
330pub fn process_responses_event(
331 event: ResponsesStreamEvent,
332) -> std::result::Result<Option<ResponseEvent>, ResponsesEventError> {
333 match event.kind.as_str() {
334 "response.output_item.done" => {
335 if let Some(item_val) = event.item {
336 if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
337 return Ok(Some(ResponseEvent::OutputItemDone(item)));
338 }
339 debug!("failed to parse ResponseItem from output_item.done");
340 }
341 }
342 "response.output_text.delta" => {
343 if let Some(delta) = event.delta {
344 return Ok(Some(ResponseEvent::OutputTextDelta(delta)));
345 }
346 }
347 "response.custom_tool_call_input.delta" => {
348 if let (Some(delta), Some(item_id)) =
349 (event.delta, event.item_id.clone().or(event.call_id.clone()))
350 {
351 return Ok(Some(ResponseEvent::ToolCallInputDelta {
352 item_id,
353 call_id: event.call_id,
354 delta,
355 }));
356 }
357 }
358 "response.reasoning_summary_text.delta" => {
359 if let (Some(delta), Some(summary_index)) = (event.delta, event.summary_index) {
360 return Ok(Some(ResponseEvent::ReasoningSummaryDelta {
361 delta,
362 summary_index,
363 }));
364 }
365 }
366 "response.reasoning_summary_text.done" => {
367 if let (Some(item_id), Some(text), Some(summary_index)) =
368 (event.item_id, event.text, event.summary_index)
369 {
370 return Ok(Some(ResponseEvent::ReasoningSummaryDone {
371 item_id,
372 text,
373 summary_index,
374 }));
375 }
376 }
377 "response.reasoning_text.delta" => {
378 if let (Some(delta), Some(content_index)) = (event.delta, event.content_index) {
379 return Ok(Some(ResponseEvent::ReasoningContentDelta {
380 delta,
381 content_index,
382 }));
383 }
384 }
385 "response.created" => {
386 if event.response.is_some() {
387 return Ok(Some(ResponseEvent::Created {}));
388 }
389 }
390 "response.failed" => {
391 if let Some(resp_val) = event.response {
392 let mut response_error = ApiError::Stream("response.failed event received".into());
393 if let Some(error) = resp_val.get("error")
394 && let Ok(error) = serde_json::from_value::<Error>(error.clone())
395 {
396 if is_context_window_error(&error) {
397 response_error = ApiError::ContextWindowExceeded;
398 } else if is_quota_exceeded_error(&error) {
399 response_error = ApiError::QuotaExceeded;
400 } else if is_usage_not_included(&error) {
401 response_error = ApiError::UsageNotIncluded;
402 } else if is_cyber_policy_error(&error) {
403 let message = cyber_policy_message(error.message);
404 response_error = ApiError::CyberPolicy { message };
405 } else if matches!(error.code.as_deref(), Some("invalid_prompt" | "bio_policy"))
406 {
407 let message = error
408 .message
409 .unwrap_or_else(|| "Invalid request.".to_string());
410 response_error = ApiError::InvalidRequest { message };
411 } else if is_server_overloaded_error(&error) {
412 response_error = ApiError::ServerOverloaded;
413 } else {
414 let delay = try_parse_retry_after(&error);
415 let message = error.message.unwrap_or_default();
416 response_error = ApiError::Retryable { message, delay };
417 }
418 }
419 return Err(ResponsesEventError::Api(response_error));
420 }
421
422 return Err(ResponsesEventError::Api(ApiError::Stream(
423 "response.failed event received".into(),
424 )));
425 }
426 "response.incomplete" => {
427 let reason = event.response.as_ref().and_then(|response| {
428 response
429 .get("incomplete_details")
430 .and_then(|details| details.get("reason"))
431 .and_then(Value::as_str)
432 });
433 let reason = reason.unwrap_or("unknown");
434 let message = format!("Incomplete response returned, reason: {reason}");
435 return Err(ResponsesEventError::Api(ApiError::Stream(message)));
436 }
437 "response.completed" => {
438 if let Some(resp_val) = event.response {
439 match serde_json::from_value::<ResponseCompleted>(resp_val) {
440 Ok(resp) => {
441 return Ok(Some(ResponseEvent::Completed {
442 response_id: resp.id,
443 token_usage: resp.usage.map(Into::into),
444 end_turn: resp.end_turn,
445 }));
446 }
447 Err(err) => {
448 let error = format!("failed to parse ResponseCompleted: {err}");
449 debug!("{error}");
450 return Err(ResponsesEventError::Api(ApiError::Stream(error)));
451 }
452 }
453 }
454 }
455 "response.output_item.added" => {
456 if let Some(item_val) = event.item {
457 if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
458 return Ok(Some(ResponseEvent::OutputItemAdded(item)));
459 }
460 debug!("failed to parse ResponseItem from output_item.added");
461 }
462 }
463 "response.reasoning_summary_part.added" => {
464 if let Some(summary_index) = event.summary_index {
465 return Ok(Some(ResponseEvent::ReasoningSummaryPartAdded {
466 summary_index,
467 }));
468 }
469 }
470 "codex.response.metadata"
471 | "response.content_part.added"
472 | "response.content_part.done"
473 | "response.custom_tool_call_input.done"
474 | "response.function_call_arguments.delta"
475 | "response.function_call_arguments.done"
476 | "response.in_progress"
477 | "response.metadata"
478 | "response.output_text.done"
479 | "response.reasoning_summary_part.done"
480 | "responsesapi.websocket_timing" => {
481 trace!("unhandled responses event: {}", event.kind);
482 }
483 kind if kind.ends_with(".delta") => {
484 trace!("unhandled responses event: {kind}");
485 }
486 _ => {
487 debug!(
488 "unhandled responses event: {:?}",
489 event.kind.chars().take(128).collect::<String>()
490 );
491 }
492 }
493
494 Ok(None)
495}
👆 带下划线的行可以点:点哪行,讲哪行。
⚠
常见误解:「流式只是 UI 效果,让用户早点看到字。」

它还决定工具什么时候能开始执行。Codex 收到一个完整的调用项,就把工具的执行排进一个「进行中」的队列,流里后面的内容还在继续到;CoreCoder 则把整条流收完才返回,工具在那之后才开始跑(它因此可以在流中途断掉时整条重发,注释里写着「此时还什么都没执行」,llm.py:184-187)。两种写法对「中途出错怎么办」的回答完全不同。

Codex:一个调用项到齐,工具就排进执行队列 codex@8cabf5a:codex-rs/core/src/session/turn.rs:2348-2360需联网 Apache-2.0
2348 let output_result =
2349 match handle_output_item_done(&mut ctx, item, previously_streamed_item)
2350 .instrument(handle_responses)
2351 .await
2352 {
2353 Ok(output_result) => output_result,
2354 Err(err) => break Err(err),
2355 };
2356 if let Some(tool_future) = output_result.tool_future {
2357 in_flight.push_back(tool_future);
2358 }
2359 if let Some(agent_message) = output_result.last_agent_message {
2360 last_agent_message = Some(agent_message);
👆 带下划线的行可以点:点哪行,讲哪行。

CoreCoder 的拼装与本课项目 miniagent/stream.py 是同一个思路:按 index 各开一个槽,参数字符串往里追加,整条流结束再逐个解析。它和本组件「按调用分别累积」这条策略做的是同一件事——只有一处不同,留到岔路 B2 再看(第 248-249 行):

CoreCoder v0.7.0 _drain:按 index 分槽,结束后解析 CoreCoder@v0.7.0:corecoder/llm.py:228-250需联网 MIT
228 # accumulate tool calls across chunks
229 if getattr(delta, "tool_calls", None):
230 for tc_delta in delta.tool_calls:
231 idx = tc_delta.index
232 if idx not in tc_map:
233 tc_map[idx] = {"id": "", "name": "", "args": ""}
234 if tc_delta.id:
235 tc_map[idx]["id"] = tc_delta.id
236 if tc_delta.function:
237 if tc_delta.function.name:
238 tc_map[idx]["name"] = tc_delta.function.name
239 if tc_delta.function.arguments:
240 tc_map[idx]["args"] += tc_delta.function.arguments
241
242 # parse accumulated tool calls
243 parsed: list[ToolCall] = []
244 for idx in sorted(tc_map):
245 raw = tc_map[idx]
246 try:
247 args = json.loads(raw["args"])
248 except (json.JSONDecodeError, KeyError):
249 args = {}
250 parsed.append(ToolCall(id=raw["id"], name=raw["name"], arguments=args))
👆 带下划线的行可以点:点哪行,讲哪行。

极小真数字:两个调用、六帧、一个时间轴

并行探针的 6 帧全在下面(毫秒是距请求发出的时刻,客户端测;耗时受本机同时运行的其他负载影响,只能粗看;token 与轮数可比):

T5 并行探针:6 帧,首帧 323.8 ms,末帧 547.3 ms(本机 mlx_lm.server,关思考)
👆 公式里带下划线的每一项都可以点——点哪项，讲哪项。
用 ported 拼装器把这 6 帧拼起来,与录制时拼好的消息逐字节一致;与同 seed 的非流式回复相比,只有服务端随机生成的调用 id 不同(组件底部自检)。

同一个任务开思考的第一轮,706 帧:首帧 125.9 ms 就到了——可那是思考;第一段正文 6,631 ms,工具调用 6,798 ms。「首 token 时间」在 agent 里至少有三种读法,用哪一种要先说清;它们的机制(prefill 与 decode)在 LLM 课 C13 第 1 课和第 5 课。

⚠
常见误解:「首 token 时间就是模型开始干活的时刻。」

在本机服务上,只调工具的轮次里首帧就是整个调用:T5 的 agent 运行中,第 1-6 轮首帧在 323–519 ms 之间,每一轮只有 4 帧;第 7 轮先流出一段正文,然后安静了 885.8 ms——模型在写一个工具调用,服务端解析失败把它丢了,这段时间一帧也没有。岔路 B1 用 span 瀑布把这 9 轮摊开看。

🔍 把 agent 全程 9 轮与 706 帧的思考摊开看
真实录制:agent 全程流式(关思考)与开思考的一轮
① 原始事件
② 参数缓冲区
③ 解析结果

第 7 轮:108 帧里,正文一段段到,然后一帧空的 delta,接着 finish_reason: tool_calls——却没有任何调用。切到第 8、9 轮:只有 4 帧,首帧要等 1 秒左右。切到「开思考一轮」:706 帧,紫色的思考帧密密排到 6.6 秒,正文和工具调用挤在最后。

证据三问

问本课的回答
① 它解决什么问题?流式回复里,工具参数分段到达;要在「拿到完整参数」的那一刻解析、执行,不早也不晚
② 为什么可能有效?共用缓冲时,两个非空 JSON 对象首尾相接一定不是合法 JSON;按调用分开累积、结束才解析,就不会拿到半截
③ 实验上呢?本机真实流:共用缓冲失败、分开累积成功、拼出的消息与录制逐字节一致;「每来一段就 parse」只在构造的云端事件流上失败

演进地图:你在这里

模型接入层的演进:每一步都被上一步的真数字逼出来
  1. 起点 写死一家的线格式
    CoreCoder(OpenAI 兼容 + litellm 兜底)、本课项目 miniagent(标准库 urllib)
    解决了最短的路:CoreCoder 的 llm.py 净 259 行,换 provider 只改 base URL 与 key。
    卡在同一段带工具调用的对话,换成 Anthropic Messages,9 个语义位置里 9 个换了路径或类型(参数从字符串变成对象,5 条消息变 3 条);Codex 干脆不认 Chat:配置 wire_api = "chat" 直接报错。
  2. 第一步 加一层抽象,或只留一种用满它
    Pi 自建 pi-ai · OpenCode 用 Vercel AI SDK · Codex 只留 Responses
    解决了Pi 把 10 种协议与 40 个 provider 做成正交的两维;OpenCode 挂了 19 个 @ai-sdk/* 依赖;Codex 的 WireApi 只剩 1 个变体。
    卡在抽象层得替每一家补差异:OpenCode 逐 provider 改写消息的 transform.ts 有 1,909 行,还给 8 个 @ai-sdk 包打了补丁;而且抽象挡不住流式与缓存的细节。
  3. 第二步 流式:按调用分别累积,结束才解析 你在这里
    Anthropic 流式文档 · OpenAI function-calling 流式指南 · Codex 只收 output_item.done
    解决了开思考的一轮,第一帧 125.9 ms 就到;整包等回复的话,正文要等到 6,631 ms、工具调用 6,798 ms。
    卡在流式只让回复来得早,不让它便宜:T5 的 9 轮每轮整份重发,输入累计 15,921 token,是最后一轮 2,389 的 6.7 倍。
  4. 第三步 缓存感知的账单 + 稳定前缀
    Anthropic / OpenAI prompt caching · Codex prompt_cache_key · OpenCode applyCaching
    解决了按 usage 四段计价:T5 在 Claude Sonnet 5 价目下 $0.0140;CoreCoder 按全价估 $0.0371,是它的 2.64 倍。
    卡在缓存只省输入:开思考的 pd-v2-unlimited-s3 账单里输出占 94%;前缀里一处变、后面全失效——T10 刷新组第 7 轮 system 重建,命中从 2,072 掉到 294。(→ A9:能跑、能算账了,怎么知道它好不好?)

数字口径:pin 源码计数(Codex 8cabf5a、OpenCode v1.18.32、Pi 31b513e3、CoreCoder v0.7.0)、项目录制 T5 / T10 与逐轮 usage(本机 Qwen3-8B,token 按它的 tokenizer 数)、官方价格页快照 2026-09-26;由 a08-model-access.json 构建时现算。

一句话

📌
流式工具参数按调用分别累积,收到结束信号才解析。
🚦

先停在这里:记住三种结束信号——Anthropic 的 content_block_stop、Responses 的 .done / output_item.done、Chat 兼容流只能等 finish_reason。
再深一层:去岔路出错、限流、变慢,怎么定位?,用 span 瀑布看这 9 轮的时间花在哪,以及流断了、被限流了各家怎么重试。

下一个缺口

每轮都重发全部历史,钱花在哪?缓存能省多少? → 第 3 课

流式让回复来得早,但不让它便宜:T5 的 9 轮每一轮都把全部历史重发一遍。每一帧 usage 里那几个数字,到底怎么变成钱?

你可能会问