大模型异步调用、中途干预与推理控制实践

一、先确认平台支持哪些能力

不同平台对异步工具调用、运行中干预和推理强度调节的支持差别很大。开始设计前，请查阅所用平台的官方文档，确认三件事：工具调用期间模型能否继续输出文本；运行中能否发送干预消息；推理强度参数的名称、取值范围，以及能否在多轮对话中修改。本文中的事件名和参数名均为示意，请以平台文档为准。

二、用系统提示词约定输出协议

在异步场景中，系统提示词的主要作用是约定工具调用与输出的协议。例如，规定工具返回错误时的格式，说明模型收到错误后应该向用户说什么，并明确哪些字段必须出现在最终回答中。应用代码因此可以稳定解析模型输出，前端也能按固定结构渲染结果。

风格要求同样可以写入提示词，比如限定回答长度，或要求直接给出结论。但系统提示词不是安全边界。多轮对话中的恶意内容仍可能让模型偏离预设方向，所以涉及权限、数据和资金的操作，必须由应用代码校验，不能只依赖提示词限制。

三、通过流式处理让工具并发运行

传统做法中，应用要等模型返回完整响应后才执行工具，工具结果再交回模型。工具耗时较长时，用户会长时间看不到任何输出。

流式接口允许应用边接收边处理，但要注意一个关键细节。工具调用的参数通常分成多个 chunk 下发，每个 chunk 只包含一段 JSON 字符串。应用需要按 index 或 call_id 把这些片段拼接起来，等到平台标记该调用完成后，才能解析 JSON 并执行工具。直接解析未闭合的片段会报错，更糟的情况是执行了一组不完整的参数。

下面的代码是示意，事件类型和字段名需替换为平台的实际名称：

import asyncio
import json
async def run_turn(client, messages, tools):
    buffers = {}   # index -> {"id", "name", "args"}
    tasks = {}     # call_id -> asyncio.Task
    outcome = {}   # call_id -> 脱敏后的结果
    async with client.stream(messages=messages, tools=tools) as stream:
        async for event in stream:
            if event.type == "text_delta":
                print(event.text, end="", flush=True)
            elif event.type == "tool_call_delta":
                buf = buffers.setdefault(event.index, {"id": None, "name": None, "args": ""})
                buf["id"] = buf["id"] or event.call_id
                buf["name"] = buf["name"] or event.name
                buf["args"] += event.arguments_fragment
            elif event.type == "tool_call_complete":
                buf = buffers.pop(event.index)
                try:
                    args = json.loads(buf["args"])  # 只在参数完整后解析
                except json.JSONDecodeError:
                    outcome[buf["id"]] = {"ok": False, "error": "invalid_arguments"}
                    continue
                tasks[buf["id"]] = asyncio.create_task(run_tool(buf["name"], args))
    results = await asyncio.gather(*tasks.values(), return_exceptions=True)
    for call_id, res in zip(tasks.keys(), results):
        if isinstance(res, BaseException):
            outcome[call_id] = {"ok": False, "error": "tool_failed"}  # 不返回堆栈
        else:
            outcome[call_id] = {"ok": True, "data": res}
    return outcome

这段代码有三个要点。第一，参数完整后才启动工具，避免执行未闭合的 JSON。第二，工具在后台并发运行，读取流的过程不会被阻塞。第三，return_exceptions=True 会把工具异常作为返回值交还。原始异常信息不能直接交给模型或用户，因为堆栈里可能包含路径、密钥或内部地址。服务端应把完整异常写入内部日志，返回给模型的只是错误码和简短说明。

工具结果返回后，应用要把结果作为工具消息追加到对话中，再发起下一轮请求。

“模型在工具运行时继续生成”取决于平台。有些平台允许在工具结果返回前继续输出过渡文本，有些平台则要求工具结果提交后才能继续推理。设计前应先实测，不能仅凭文档推断。

四、运行中途修改指令

用户常在模型输出过程中改变想法。要支持这一点，应用与服务端之间需要一条能在生成中途传递消息的通道。

WebSocket 是双向通道，客户端可以直接发送干预消息。如果输出通过 SSE 接收，SSE 只能从服务端向客户端推送，干预就要通过另一条普通 HTTP 请求发出，并携带当前响应的标识，例如 response_id 或 trace_id。

使用 SSE 时，取消流程通常分三步。前端用 AbortController 中止读取流，停止接收新内容。随后向服务端发送取消或干预请求，并附上 response_id 和幂等键。最后等待服务端确认。只关闭连接不一定能停止服务端的生成，有些平台会继续计算并计费，因此应显式发送取消请求。

服务端收到干预后，通常会在下一个安全的生成边界（例如当前 token 输出完成后）停止本次输出。已生成的部分保留下来，作为助手消息的前缀，再结合新指令开启新的生成。

实现时要处理以下边界条件：

1. 干预消息晚于响应结束才到达。服务端应返回明确的”响应已完成”状态，不能静默丢弃。
2. 同一响应收到多条干预。可以按到达顺序合并，也可以只保留最后一条。无论选哪种，都要在接口文档中写明规则。
3. 工具仍在运行时收到干预。应取消或标记挂起的工具调用，并丢弃其过期结果，避免旧结果污染新回答。
4. 客户端断线。服务端应设置超时并释放资源，同时决定是否保存已生成的部分输出。
5. 网络重试导致重复提交。干预请求应携带幂等键，确保同一指令只执行一次。

五、动态调整推理预算并复用缓存

长对话中，简单问题和复杂问题对推理的需求差异很大。固定的推理强度会造成资源浪费，或让复杂任务的质量不足。按任务难度分配推理预算，可以把算力用在需要它的地方。

下面是请求结构的示意，字段名以平台文档为准：

{
  "model": "your-model-name",
  "reasoning": {
    "effort": "medium"
  },
  "messages": []
}

不同厂商的参数形式并不相同。以下是截至撰写时的公开文档中的常见写法，使用前请核对当前版本：

• Anthropic 的扩展思考通过 thinking: {"type": "enabled", "budget_tokens": N} 设置预算，文档要求 budget_tokens 不低于 1024。
• OpenAI 的推理模型在 Chat Completions 中使用 reasoning_effort，取值如 low、medium、high。Responses API 则使用 reasoning: {"effort": ...} 结构。

分档时，不要照抄固定数值。扫描目录、整理文件清单这类机械任务，可以使用平台允许的最低档位，或在平台支持时关闭推理。分析性能瓶颈、给出重构方案这类复杂任务，则提高到压测后确定的档位。如果预算低于平台的最小值，请求可能被拒绝，或被平台自动取整，因此分档前要先确认平台的下限。

缓存复用有几个前提。第一，前缀必须完全一致，包括系统提示词、历史消息和工具定义。多一个空格或换一个字段顺序，都可能导致未命中。第二，模型名称和部署版本要相同。第三，推理参数的改变有可能影响提示模板的渲染结果，进而使缓存失效，调整预算后需要重新观测命中情况。第四，缓存有生存时间，长时间空闲后会被回收。

实践中，应把稳定内容放在请求前部，把动态内容（如当前时间、用户信息）放在末尾。命中效果要通过平台返回的用量字段验证，字段名以平台文档为准。

六、生产环境的超时、限流与成本控制

异步工具调用会增加并发。若不加限制，可能触发平台限流，或因重试而让费用超出预期。建议做到以下几点：

• 为每个工具设置超时，超时后向模型返回明确的错误信息。
• 限制单轮最大并发工具数。
• 对可重试的错误使用指数退避，并设置最大重试次数。
• 为每次对话设置 token 预算上限，并持续监控推理 token 的消耗。
• 工具异常日志要脱敏，过滤密钥、文件路径和用户数据。堆栈信息只写入内部日志，不进入模型的下一轮上下文。

基础设施方面，消息队列、无服务器函数和容器平台都可以承接耗时的工具任务。选型时应关注冷启动延迟、最长执行时间和并发配额，这些指标直接决定异步链路是否稳定。

七、底层特性选型检查清单

• 平台是否支持工具调用期间继续输出？是否已通过实测验证？
• 运行中干预使用哪种通道？断线和重复提交如何处理？
• 推理参数的名称、取值范围、最小预算和计费方式是什么？
• 缓存前缀是否稳定，命中率能否观测？
• 工具是否设置了超时、并发上限和重试策略？
• 权限、数据和资金相关操作是否由应用代码校验？

八、常见问题

异步调用一定能加快响应吗？

不一定。工具本身执行很快，或平台要求工具结果返回后才能继续推理时，收益有限。应以实测延迟为准，比较改造前后的首字延迟和整轮耗时。

缓存复用失败的常见原因有哪些？

常见原因包括：系统提示词中混入了时间戳或随机 ID，前缀每次都不同；工具定义的顺序发生变化；模型版本或推理参数被修改；缓存超过生存时间后被回收。排查时，先对比两次请求的前缀是否逐字节一致，再查看平台返回的缓存命中字段。