> ## Documentation Index > Fetch the complete documentation index at: https://docs.qoder.com/llms.txt > Use this file to discover all available pages before exploring further. # 增量流式 > 为 Cloud Agent Session 开启并验证增量流式事件。 增量流式让客户端在最终完整 `agent.message` 事件之前，提前收到 assistant 输出片段。创建 Session 时开启： ```json theme={null} { "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 元字段。 顶层增量事件类型： ```text theme={null} 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` 事件返回 | 常见事件形态： ```json theme={null} { "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 } } } ``` ```json theme={null} { "type": "agent.content_block_delta", "message_id": "asst_...", "index": 0, "delta": { "type": "text_delta", "text": "Hello" } } ``` ```json theme={null} { "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。 ```bash theme={null} 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： ```bash theme={null} 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}' ``` 预期响应： ```json theme={null} { "id": "sess_...", "incremental_streaming_enabled": true, "status": "idle" } ``` 在一个终端打开 Session SSE stream： ```bash theme={null} 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 ``` 在另一个终端发送用户消息： ```bash theme={null} 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 中应该在最终完整事件前看到增量序列： ```text theme={null} 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 ``` 验证历史回放： ```bash theme={null} 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： ```bash theme={null} 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 使用同样的解析循环： ```bash theme={null} 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。响应中应包含： ```json theme={null} { "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 重连。