增量流式 - Qoder

增量流式让客户端在最终完整 agent.message 事件之前，提前收到 assistant 输出片段。创建 Session 时开启：

{
  "incremental_streaming_enabled": true
}

该配置会持久化在 Session 上，不通过 stream 请求的 query 参数或 header 控制。

接口

接口路径不变：

API	增量行为
`GET /api/v1/cloud/sessions/{session_id}/events/stream`	Session 级实时 SSE 流
`GET /api/v1/cloud/sessions/{session_id}/threads/{thread_id}/stream`	thread 级实时 SSE 流
`GET /api/v1/cloud/sessions/{session_id}/events`	Session 级历史事件回放
`GET /api/v1/cloud/sessions/{session_id}/threads/{thread_id}/events`	thread 级历史事件回放

当 incremental_streaming_enabled 为 false 或省略时，这些接口保持原有行为，不返回增量事件。

事件模型

公开事件与 qodercli/Anthropic raw stream event 对齐：CAW 侧产生的 raw stream event 类型为 message_start、content_block_start 等；CAS 对外暴露时给事件 type 加上 agent. 前缀，并补充 id、session_id、session_thread_id、turn_id、message_id、parent_tool_use_id、processed_at 等 CAS 元字段。顶层增量事件类型：

agent.message_start
agent.content_block_start
agent.content_block_delta
agent.content_block_stop
agent.message_delta
agent.message_stop

具体片段类型位于 agent.content_block_delta.delta.type，不是顶层事件类型。当前实现中，agent.content_block_start.content_block.type 可能为：

`content_block.type`	说明
`thinking`	thinking block，初始 `thinking` 为空字符串
`text`	text block，初始 `text` 为空字符串
`tool_use`	工具调用 block，包含 `id`、`name` 和初始空 `input`

`delta.type`	含义
`text_delta`	文本输出片段
`thinking_delta`	模型/provider 输出 thinking 时的思考片段
`signature_delta`	thinking block 的签名片段，存在时透出
`input_json_delta`	工具入参 JSON 片段。产品语义里的 `tool_input_delta` 对外使用这个 wire shape，字段为 `partial_json`
`tool_output_delta`	预留给未来的工具输出流式；当前实现不会产生该 delta，工具结果仍通过完整 `agent.tool_result` 事件返回

常见事件形态：

{
  "type": "agent.message_start",
  "message_id": "asst_...",
  "message": {
    "id": "asst_...",
    "type": "message",
    "role": "assistant",
    "model": "qwen3-coder-plus",
    "content": [],
    "stop_reason": null,
    "stop_sequence": null,
    "usage": {
      "input_tokens": 0,
      "output_tokens": 0
    }
  }
}

{
  "type": "agent.content_block_delta",
  "message_id": "asst_...",
  "index": 0,
  "delta": {
    "type": "text_delta",
    "text": "Hello"
  }
}

{
  "type": "agent.message_delta",
  "message_id": "asst_...",
  "delta": {
    "stop_reason": "end_turn",
    "stop_sequence": null,
    "container": null
  },
  "usage": {
    "input_tokens": 123,
    "output_tokens": 45
  }
}

完整的 agent.message 仍会在增量序列之后返回，作为最终权威结果。对于同一个 text block，按顺序拼接 text_delta.text 应等于最终 agent.message.content[index].text；如果 provider 漏发了最后一小段文本，CAW 会在 agent.content_block_stop 前补发尾部 text_delta。

快速验证

前置条件：

有效 PAT，写入 QODER_PAT
已存在的 Agent ID，写入 AGENT_ID
已存在的 Environment ID，写入 ENVIRONMENT_ID
本地已安装 jq

设置 API base。生产环境默认使用 api.qoder.com；验证预发部署时可切到 Global Test base。

export BASE_URL="https://api.qoder.com/api/v1/cloud"
# export BASE_URL="https://test-api.qoder.ai/api/v1/cloud"

export QODER_PAT="pat_..."
export AGENT_ID="agent_..."
export AGENT_VERSION="1"
export ENVIRONMENT_ID="env_..."

创建开启增量流式的 Session：

SESSION_JSON=$(
  jq -n \
    --arg agent_id "$AGENT_ID" \
    --argjson agent_version "$AGENT_VERSION" \
    --arg environment_id "$ENVIRONMENT_ID" \
    '{
      agent: {id: $agent_id, type: "agent", version: $agent_version},
      environment_id: $environment_id,
      title: "incremental streaming verification",
      incremental_streaming_enabled: true
    }' |
  curl -s -X POST "$BASE_URL/sessions" \
    -H "Authorization: Bearer $QODER_PAT" \
    -H "Content-Type: application/json" \
    --data-binary @-
)

export SESSION_ID=$(echo "$SESSION_JSON" | jq -r '.id')
echo "$SESSION_JSON" | jq '{id, incremental_streaming_enabled, status}'

预期响应：

{
  "id": "sess_...",
  "incremental_streaming_enabled": true,
  "status": "idle"
}

在一个终端打开 Session SSE stream：

curl -sN "$BASE_URL/sessions/$SESSION_ID/events/stream" \
  -H "Authorization: Bearer $QODER_PAT" \
  -H "Accept: text/event-stream" |
while IFS= read -r line; do
  case "$line" in
    event:*) echo "$line" ;;
    data:*) echo "${line#data: }" | jq -cr '{type, message_id, index, block_type: .content_block.type, delta_type: .delta.type, text: .delta.text, thinking: .delta.thinking, partial_json: .delta.partial_json}' ;;
  esac
done

在另一个终端发送用户消息：

curl -s -X POST "$BASE_URL/sessions/$SESSION_ID/events" \
  -H "Authorization: Bearer $QODER_PAT" \
  -H "Content-Type: application/json" \
  -d '{
    "events": [
      {
        "type": "user.message",
        "content": [
          {"type": "text", "text": "Reply with exactly one short sentence."}
        ]
      }
    ]
  }' | jq

stream 中应该在最终完整事件前看到增量序列：

event: agent.message_start
event: agent.content_block_start
event: agent.content_block_delta
{"type":"agent.content_block_delta","delta_type":"text_delta","text":"..."}
event: agent.content_block_stop
event: agent.message_delta
event: agent.message_stop
event: agent.message
event: session.status_idle

验证历史回放：

curl -s "$BASE_URL/sessions/$SESSION_ID/events?limit=200" \
  -H "Authorization: Bearer $QODER_PAT" |
jq -r '.data[].type' |
grep -E 'agent\.message_start|agent\.content_block_delta|agent\.message_stop'

验证 thread 维度回放和 stream：

export THREAD_ID=$(
  curl -s "$BASE_URL/sessions/$SESSION_ID/threads?limit=20" \
    -H "Authorization: Bearer $QODER_PAT" |
  jq -r '.data[0].id'
)

curl -s "$BASE_URL/sessions/$SESSION_ID/threads/$THREAD_ID/events?limit=200" \
  -H "Authorization: Bearer $QODER_PAT" |
jq -r '.data[].type' |
grep -E 'agent\.message_start|agent\.content_block_delta|agent\.message_stop'

如果要查看 thread 维度实时流，对 thread stream endpoint 使用同样的解析循环：

curl -sN "$BASE_URL/sessions/$SESSION_ID/threads/$THREAD_ID/stream" \
  -H "Authorization: Bearer $QODER_PAT" \
  -H "Accept: text/event-stream" |
while IFS= read -r line; do
  case "$line" in
    event:*) echo "$line" ;;
    data:*) echo "${line#data: }" | jq -cr '{type, message_id, index, block_type: .content_block.type, delta_type: .delta.type, text: .delta.text, thinking: .delta.thinking, partial_json: .delta.partial_json}' ;;
  esac
done

关闭开关对照

再创建一个不传 incremental_streaming_enabled 或显式传 false 的 Session。响应中应包含：

{
  "incremental_streaming_enabled": false
}

发送同样的 user.message 后，stream 和 history 应包含 agent.message、session.status_idle 等完整事件，但不应包含上面的增量事件类型。

解析检查项

将 SSE event: 和 JSON data.type 当作公开事件类型。
对于 delta.type == "text_delta" 的 agent.content_block_delta，追加 delta.text 来重建文本。
对于 delta.type == "thinking_delta" 的 agent.content_block_delta，追加 delta.thinking 来重建 thinking；后续仍可能收到完整兼容事件 agent.thinking。
对于工具入参增量，判断 delta.type == "input_json_delta" 并追加 delta.partial_json；不要期待顶层 tool_input_delta 事件。
当前不会收到 tool_output_delta；工具执行结果以完整 agent.tool_result 返回。
使用 index 区分多个 content block。
processed_at 在 agent 生成事件上可能缺失，解析时按可选字段处理。
如果客户端支持同一连接多轮对话，收到 session.status_idle 后继续保持连接。
网络中断后用 Last-Event-ID 携带最后收到的事件 ID 重连。

​接口

​事件模型

​快速验证

​关闭开关对照

​解析检查项

接口

事件模型

快速验证

关闭开关对照

解析检查项