Chat Completions 迁移 Responses API：函数调用完整循环与验收

import OpenAI from "openai";
import { toResponseInputItems } from "openai/lib/responses/ResponseInputItems";

const toolName = "lookup_fulfillment_status";
const tools = [{
  type: "function", name: toolName,
  description: "查询已获授权订单的发货和运输状态，只读",
  strict: true,
  parameters: {
    type: "object",
    properties: { order_ref: { type: "string" } },
    required: ["order_ref"], additionalProperties: false,
  },
}];
const fulfillment = new Map([["CN-7042", {
  warehouse_status: "shipped", carrier_status: "in_transit",
  last_scan: "杭州转运中心",
}]]);

export async function queryFulfillment({
  create, toInputItems, model, authorizedOrders,
  maxRounds = 6, budgetMs = 30000,
}) {
  if (!model || !(authorizedOrders instanceof Set)) {
    throw new Error("缺少模型或后端授权范围");
  }
  const input = [{ role: "user", content: "订单 CN-7042 发货了吗？到哪里了？" }];
  const instructions = "先查询订单，再根据工具结果回答；查询失败时不可猜测物流。";
  const deadline = Date.now() + budgetMs;
  const ledger = new Map();
  const audit = [];
  const checkBudget = () => {
    if (Date.now() >= deadline) throw new Error("应用处理时间已用尽");
  };

  async function dispatch(call) {
    try {
      if (call.name !== toolName) throw new Error("未知函数");
      const args = JSON.parse(call.arguments);
      if (!args || Array.isArray(args) || typeof args !== "object" ||
          Object.keys(args).length !== 1 ||
          typeof args.order_ref !== "string" || !/^CN-\d{4}$/.test(args.order_ref)) {
        throw new Error("参数必须只含合法 order_ref");
      }
      if (!authorizedOrders.has(args.order_ref)) throw new Error("无权查询此订单");
      const data = fulfillment.get(args.order_ref);
      if (!data) return { ok: false, code: "not_found" };
      // 这里只读本地假数据；业务接口应在同一授权检查之后调用。
      return { ok: true, data };
    } catch {
      return { ok: false, code: "tool_rejected" };
    }
  }

  for (let round = 0; round < maxRounds; round += 1) {
    checkBudget();
    const response = await create({
      model, instructions, input: [...input], tools,
      tool_choice: round === 0 ? { type: "function", name: toolName } : "auto",
      parallel_tool_calls: false, store: false,
      include: ["reasoning.encrypted_content"],
    });
    checkBudget();
    if (response.status !== "completed") {
      throw new Error(`响应未完成: ${response.status}; ${JSON.stringify(
        response.incomplete_details ?? response.error ?? null
      )}`);
    }
    if (typeof response.id !== "string" || !Array.isArray(response.output)) {
      throw new Error("响应结构缺失");
    }
    const calls = new Map();
    for (const item of response.output) {
      if (!["message", "reasoning", "function_call"].includes(item.type)) {
        throw new Error(`此示例未实现输出类型: ${item.type}`);
      }
      if (item.type !== "function_call") continue;
      if (typeof item.call_id !== "string" || !item.call_id ||
          typeof item.name !== "string" || typeof item.arguments !== "string" ||
          (item.status && item.status !== "completed")) {
        throw new Error("函数调用缺少有效身份或未完成");
      }
      const signature = JSON.stringify([item.name, item.arguments]);
      const prior = calls.get(item.call_id);
      if (prior && prior.signature !== signature) throw new Error("同一 call_id 内容冲突");
      calls.set(item.call_id, { item, signature });
    }

    if (calls.size === 0) {
      const parts = [];
      for (const item of response.output) {
        if (item.type !== "message") continue;
        for (const part of item.content ?? []) {
          if (part.type === "refusal") throw new Error("模型拒绝，未交付查询回答");
          if (part.type === "output_text" && item.role === "assistant" &&
              (!item.phase || item.phase === "final_answer")) parts.push(part.text);
        }
      }
      const text = parts.join("\n").trim();
      if (!text) throw new Error("没有完整最终文本");
      if (!audit.some((entry) => entry.ok)) throw new Error("没有成功查询，不能验收业务成功");
      return { text, audit };
    }

    // 官方 helper 转换所有相关输出，保留推理项和消息 phase。
    input.push(...toInputItems(response.output));
    for (const { item: call, signature } of calls.values()) {
      checkBudget();
      const key = JSON.stringify([response.id, call.call_id]);
      let entry = ledger.get(key);
      if (entry && entry.signature !== signature) throw new Error("重放调用内容冲突");
      if (!entry) {
        // 先登记待执行 Promise；dispatch 在微任务中执行，避免并发重入窗口。
        entry = { signature, pending: Promise.resolve().then(() => dispatch(call)) };
        ledger.set(key, entry);
      }
      const result = await entry.pending;
      audit.push({ response_id: response.id, call_id: call.call_id, ok: result.ok });
      input.push({ type: "function_call_output", call_id: call.call_id,
        output: JSON.stringify(result) });
    }
  }
  throw new Error("调用轮数已用尽，未取得最终回答");
}

const model = process.env.OPENAI_MODEL;
if (!model) throw new Error("请设置已验证的 OPENAI_MODEL");
const client = new OpenAI();
const result = await queryFulfillment({
  model, authorizedOrders: new Set(["CN-7042"]),
  create: (request) => client.responses.create(request),
  toInputItems: toResponseInputItems,
});
console.log(result.text);
console.log(JSON.stringify(result.audit));

npm install openai
node responses-fulfillment.mjs

export async function collectCompletedResponse(events) {
  const seen = new Map();
  const items = new Map();
  const argumentsByItem = new Map();
  let completed;
  for await (const event of events) {
    if (!Number.isInteger(event.sequence_number)) throw new Error("事件缺少序号");
    const fingerprint = JSON.stringify(event);
    if (seen.has(event.sequence_number)) {
      if (seen.get(event.sequence_number) !== fingerprint) throw new Error("事件序号冲突");
      continue;
    }
    seen.set(event.sequence_number, fingerprint);
    if (["response.failed", "response.incomplete", "response.cancelled"].includes(event.type)) {
      throw new Error(`流式响应失败: ${event.type}`);
    }
    if (event.type === "error") throw new Error("流式传输返回错误");
    if (event.type === "response.output_item.added" ||
        event.type === "response.output_item.done") {
      items.set(event.output_index, event.item);
    } else if (event.type === "response.function_call_arguments.delta") {
      argumentsByItem.set(event.item_id,
        (argumentsByItem.get(event.item_id) ?? "") + event.delta);
    } else if (event.type === "response.function_call_arguments.done") {
      argumentsByItem.set(event.item_id, event.arguments);
    } else if (event.type === "response.completed") {
      completed = event.response;
      if (completed.status !== "completed" || !Array.isArray(completed.output)) {
        throw new Error("completed 事件缺少完整 Response");
      }
      for (const [index, item] of items) {
        if (item.type !== "function_call") continue;
        const finalItem = completed.output[index];
        if (!finalItem || finalItem.id !== item.id || finalItem.call_id !== item.call_id) {
          throw new Error("函数输出项身份不匹配");
        }
        const args = argumentsByItem.get(item.id);
        if (args !== undefined && args !== finalItem.arguments) {
          throw new Error("函数参数完成值不匹配");
        }
      }
      return completed;
    }
  }
  throw new Error("事件流提前结束，未收到 response.completed");
}