两种组装方式
| 阶段 | Chat Completions | Responses |
|---|---|---|
| 关联分片 | choices[0].delta.tool_calls[].index | output_index 与 item_id |
| 参数增量 | function.arguments 字符串 | response.function_call_arguments.delta 的 delta |
| 调用身份 | id 与 function.name | output item 的 id、call_id、name |
| 结束验证 | finish_reason: tool_calls，且迭代无异常结束 | arguments.done、output_item.done、response.completed 相互一致，且迭代无异常结束 |
| 回传结果 | role: tool 与 tool_call_id | type: function_call_output 与 call_id |
[DONE] 字符串。
运行准备
按 TypeScript或 Python安装已有 OpenAI SDK；无需新增依赖。预先设置网关令牌API_KEY，然后设置含 /v1 的 Base URL：
export API_BASE_URL="https://gateway.example.com/v1"
export API_MODEL="YOUR_TOOL_CAPABLE_MODEL_ID"
# API_KEY must already be set in the process environment.
node examples/stream-tools.mts 或 .venv/bin/python examples/stream-responses-tools.py。单独保存下方代码时可运行 node stream-tools.mts 或 python stream-responses-tools.py。例子请求计算 7 × 8；最终措辞由模型决定，不强制输出某个固定句子。
TypeScript：Chat Completions
stream-tools.mts
import OpenAI from 'openai';
const { API_BASE_URL, API_KEY, API_MODEL } = process.env;
if (!API_BASE_URL || !API_KEY || !API_MODEL) {
throw new Error('Set API_BASE_URL, API_KEY, and API_MODEL first.');
}
const client = new OpenAI({ baseURL: API_BASE_URL.replace(/\/+$/, ''), apiKey: API_KEY,
timeout: 60_000, maxRetries: 0 });
const messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[] = [
{ role: 'user', content: 'What is 7 times 8? Use multiply.' },
];
const tools: OpenAI.Chat.Completions.ChatCompletionTool[] = [{
type: 'function', function: { name: 'multiply', description: 'Multiply two finite numbers.', strict: true,
parameters: { type: 'object', properties: { a: { type: 'number' }, b: { type: 'number' } },
required: ['a', 'b'], additionalProperties: false } },
}];
const seenIds = new Set<string>();
try {
for (let round = 0; round < 3; round++) {
const calls = new Map<number, { id: string; type: 'function'; function: { name: string; arguments: string } }>();
let finish: string | null = null;
let text = '';
let events = 0;
const stream = await client.chat.completions.create({ model: API_MODEL, messages, tools, stream: true });
for await (const chunk of stream) {
if (++events > 512) throw new Error('Stream event limit reached.');
for (const choice of chunk.choices) {
if (choice.index !== 0 || finish !== null) throw new Error('Unexpected choice or event after completion.');
if (choice.delta.refusal || choice.delta.function_call) throw new Error('Refusal or unsupported legacy call.');
text += choice.delta.content ?? '';
if (text.length > 100_000) throw new Error('Text limit reached.');
for (const delta of choice.delta.tool_calls ?? []) {
if (!Number.isInteger(delta.index) || delta.index < 0 || delta.index >= 4) throw new Error('Invalid call index.');
let call = calls.get(delta.index);
if (!call) {
call = { id: '', type: 'function', function: { name: '', arguments: '' } };
calls.set(delta.index, call);
}
if (delta.type && delta.type !== 'function') throw new Error('Unsupported tool type.');
if (delta.id) {
if (call.id && call.id !== delta.id) throw new Error('Call ID changed.');
call.id = delta.id;
}
if (delta.function?.name) {
if (call.function.name && call.function.name !== delta.function.name) throw new Error('Tool name changed.');
call.function.name = delta.function.name;
}
call.function.arguments += delta.function?.arguments ?? '';
if (call.function.arguments.length > 16_384) throw new Error('Arguments limit reached.');
}
if (choice.finish_reason !== null) finish = choice.finish_reason;
}
}
// No function executes until the whole stream has ended with a valid finish reason.
if (calls.size === 0) {
if (finish !== 'stop' || !text.trim()) throw new Error('No complete answer.');
console.log(text);
break;
}
if (finish !== 'tool_calls' || round === 2) throw new Error('Incomplete calls or round limit reached.');
const ordered = [...calls.entries()].sort(([a], [b]) => a - b).map(([, call]) => call);
const outputs: OpenAI.Chat.Completions.ChatCompletionToolMessageParam[] = [];
for (const call of ordered) {
if (!call.id.trim() || seenIds.has(call.id) || call.function.name !== 'multiply') throw new Error('Invalid tool call.');
const args = JSON.parse(call.function.arguments);
if (!args || Array.isArray(args) || Object.keys(args).length !== 2
|| !Number.isFinite(args.a) || !Number.isFinite(args.b)) throw new Error('Invalid arguments.');
const value = args.a * args.b;
if (!Number.isFinite(value)) throw new Error('Invalid result.');
seenIds.add(call.id);
outputs.push({ role: 'tool', tool_call_id: call.id, content: JSON.stringify({ value }) });
}
messages.push({ role: 'assistant', content: text || null, tool_calls: ordered }, ...outputs);
}
} catch (error) {
console.error(error instanceof OpenAI.APIError && error.status
? `Request failed (HTTP ${error.status}). No automatic retry.`
: 'Streaming tool workflow failed; no incomplete answer published.');
process.exitCode = 1;
}
Python：Responses
stream-responses-tools.py
import json
import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError, APIError
history = [{"role": "user", "content": "What is 7 times 8? Use multiply."}]
tools = [{
"type": "function", "name": "multiply", "description": "Multiply two finite numbers.",
"strict": True,
"parameters": {
"type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"], "additionalProperties": False,
},
}]
seen_call_ids = set()
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
try:
for round_index in range(3):
result = None
pending = {}
with client.responses.create(
model=os.environ["API_MODEL"], input=history, tools=tools,
instructions="Use multiply for arithmetic, then answer briefly.",
store=False, include=["reasoning.encrypted_content"], stream=True,
) as stream:
for event_count, event in enumerate(stream, start=1):
if event_count > 512 or result is not None:
raise ValueError("Event limit or event after completion.")
if event.type in ("error", "response.failed", "response.incomplete",
"response.refusal.delta", "response.refusal.done"):
raise ValueError("Failed or refused response.")
if event.type == "response.output_item.added" and event.item.type == "function_call":
index = event.output_index
if type(index) is not int or index < 0 or index in pending or len(pending) >= 4:
raise ValueError("Invalid function item index.")
if (not event.item.id or not event.item.call_id
or not isinstance(event.item.arguments, str)):
raise ValueError("Missing function identity or arguments.")
pending[index] = {"id": event.item.id, "call_id": event.item.call_id,
"name": event.item.name, "arguments": event.item.arguments,
"arguments_done": False, "item_done": False}
if len(event.item.arguments) > 16_384:
raise ValueError("Arguments limit reached.")
elif event.type in ("response.function_call_arguments.delta", "response.function_call_arguments.done"):
call = pending[event.output_index]
if call["id"] != event.item_id or call["arguments_done"]:
raise ValueError("Unknown, changed, or finished function item.")
if event.type.endswith(".delta"):
call["arguments"] += event.delta
if len(call["arguments"]) > 16_384:
raise ValueError("Arguments limit reached.")
else:
if event.arguments != call["arguments"]:
raise ValueError("Final arguments differ from streamed arguments.")
call["arguments_done"] = True
elif event.type == "response.output_item.done" and event.item.type == "function_call":
call = pending[event.output_index]
if (not call["arguments_done"] or call["item_done"]
or any(getattr(event.item, key) != call[key]
for key in ("id", "call_id", "name", "arguments"))):
raise ValueError("Incomplete or inconsistent function item.")
call["item_done"] = True
elif event.type == "response.completed":
result = event.response
if result is None:
raise ValueError("Stream ended without response.completed.")
final_indices = set()
for index, item in enumerate(result.output):
if item.type != "function_call":
continue
call = pending[index]
if (not call["item_done"] or any(getattr(item, key) != call[key]
for key in ("id", "call_id", "name", "arguments"))):
raise ValueError("Completed response differs from streamed calls.")
final_indices.add(index)
if final_indices != set(pending):
raise ValueError("Completed response lost a streamed call.")
# Execute only after the completed response and the entire stream validate.
if result.status != "completed":
raise ValueError("Response did not complete.")
calls = []
for item in result.output:
if item.type == "function_call":
if getattr(item, "status", None) not in (None, "completed"):
raise ValueError("Function call did not complete.")
calls.append(item)
elif item.type == "message":
if item.status != "completed" or any(part.type != "output_text" for part in item.content):
raise ValueError("Refusal or incomplete message.")
elif item.type != "reasoning":
raise ValueError("Unsupported output item.")
if not calls:
if not result.output_text.strip():
raise ValueError("No final text.")
print(result.output_text)
break
if round_index == 2 or len(calls) > 4:
raise ValueError("Tool round or call limit reached.")
outputs = []
for call in calls:
if (call.name != "multiply" or not isinstance(call.call_id, str)
or not call.call_id.strip() or call.call_id in seen_call_ids):
raise ValueError("Unknown tool or invalid call ID.")
args = json.loads(call.arguments)
if not isinstance(args, dict) or set(args) != {"a", "b"}:
raise ValueError("Invalid tool arguments.")
if any(type(value) not in (int, float) or not math.isfinite(value) for value in args.values()):
raise ValueError("Expected finite numbers.")
value = args["a"] * args["b"]
if not math.isfinite(value):
raise ValueError("Invalid tool result.")
seen_call_ids.add(call.call_id)
outputs.append({
"type": "function_call_output", "call_id": call.call_id,
"output": json.dumps({"value": value}, allow_nan=False),
})
history.extend(item.model_dump(exclude_none=True) for item in result.output)
history.extend(outputs)
except APIStatusError as error:
raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
except APIConnectionError:
raise SystemExit("Connection failed. No automatic retry.")
except (ValueError, TypeError, AttributeError, KeyError, OverflowError, APIError):
raise SystemExit("Streaming tool workflow failed; no incomplete answer published.")
执行与失败策略
- 两个脚本都先等完整一轮结束，再验证工具白名单、关联 ID 和全部参数；仅实现无外部副作用的
multiply。严格 Schema 不能替代本地校验。 - 每轮最多 4 个调用、512 个事件；最多 3 次模型请求，第三次必须产生最终文本。参数长度上限为 16,384（TypeScript 按 UTF-16 code unit，Python 按字符）；Chat 文本缓冲最多 100,000 code unit。这些是示例限制，不是服务端限额，也不限制 SDK 解码前单个事件的字节大小。
- Responses 示例核对增量参数、done 参数和最终 output，重放完整输出项（含 reasoning / encrypted_content / phase）及结果。
store: false不依赖响应 ID 续接；目标渠道仍需支持对应字段保留。 - 两个脚本缓冲最终文本，流结束并通过完成校验后才打印。HTTP 401/429、事件级错误、拒答、截断、缺少完成标记、未知工具、重复 ID、非法参数或上限触发时退出非零，不发送下一轮，也不自动重试。
本地验证与协议范围
npm run check:examples 用本地 HTTP fixture 执行这些文件，检查交错调用、UTF-8 跨网络块、完整历史与工具结果关联、第二轮请求，以及异常时没有后续请求。模拟响应不能证明真实模型一定选择工具或目标渠道保留所有事件；上线前须用获授权的测试模型验证。
Claude 原生与 Gemini 原生使用不同内容块、事件及结果结构；本页脚本不适用于这两种原生协议。
参考：OpenAI 函数调用、Responses 流式事件。
原生协议的可运行 SSE 与工具闭环见 Claude和 Gemini，包含完成校验与签名保留。
Read in English