异步工具调用、中途引导与推理动态切换的工程实践

一、背景：从一问一答到长时任务

早期的大模型应用多是单轮问答。用户提问，模型一次性生成答案，交互边界清楚。随着智能体（agent）逐渐承担代码检索、测试执行、文档整理等多步骤工作，原有的交互方式开始暴露问题。任务可能持续数分钟，中间要调用多个外部工具，用户还会在过程中改变想法。此时，模型能否并行处理工具结果、能否接受中途修正、能否按任务难度调整计算投入，直接决定了体验与成本。

本文讨论三个方向。第一，工具调用从阻塞式走向异步式。第二，任务执行过程中允许用户插话并修正方向。第三，在长会话中按需切换推理强度，同时尽量复用已计算的上下文缓存。三者都属于通用的工程问题，与具体厂商无关，理解其机制后，可以在任意支持类似能力的服务上套用同样的设计思路。

二、异步工具调用

2.1 同步阻塞的瓶颈

传统的工具调用流程是严格串行的。模型输出一个工具调用请求，应用执行该工具，等待结果返回，再把结果交给模型继续生成。在这个过程中，模型处于等待状态，无法利用这段时间规划后续步骤或处理其他分支。

应用程序本身通常可以用线程池或协程并发执行多个工具，但如果模型层面仍然要求每次调用都等待返回，整体吞吐量仍然受限。例如，检索三个文件、运行一组测试、查询一个配置项，三件事彼此无依赖，却只能排队执行，延迟会线性累加。

2.2 异步模式的执行流程

异步工具调用的核心是把”发起调用”和”消费结果”拆开。模型发出调用后，可以继续生成与该结果无关的内容，应用在结果就绪时把它回填到上下文中。流程大致如下：

1. 模型输出若干工具调用请求，每个请求带有唯一标识。
2. 应用并发执行这些请求，不阻塞模型输出。
3. 某个结果返回后，应用以标识为键将其写入上下文。
4. 模型在合适的时机读取已返回的结果，继续推理。

这种模式的收益主要体现在长链路任务中。若某个工具耗时较长，模型可以先处理已经就绪的结果，或先整理其他步骤的思路，整体等待时间缩短。

下面用协程演示应用侧如何并发调度工具并逐个回填结果。示例中的工具函数是占位实现，实际场景中可替换为对象存储查询、构建任务或数据库访问等通用能力。

import asyncio
async def call_tool_async(name: str, args: dict) -> dict:
    # 占位实现：模拟一次外部调用耗时
    await asyncio.sleep(0.5)
    return {"call_name": name, "status": "ok", "data": f"{name} 的结果"}
async def run_pending_calls(calls: list[tuple[str, dict]], context: list):
    pending = {asyncio.create_task(call_tool_async(n, a)) for n, a in calls}
    while pending:
        done, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
        for task in done:
            result = task.result()
            # 结果就绪即回填，模型可基于已完成部分继续推理
            context.append(result)
if __name__ == "__main__":
    ctx: list = []
    calls = [
        ("list_files", {"path": "src"}),
        ("run_tests", {"suite": "unit"}),
        ("read_config", {"key": "timeout"}),
    ]
    asyncio.run(run_pending_calls(calls, ctx))
    print(ctx)

2.3 工程注意事项

异步带来并发，也带来新的风险。需要关注以下几点。

• 结果乱序：返回顺序与发起顺序不一致，回填时必须依赖唯一标识，而不是位置。
• 超时与取消：每个调用应设置超时。超时后要明确告知模型该结果缺失，避免它基于假设继续推理。
• 幂等性：若同一调用可能被重试，写操作需要幂等设计，防止重复执行造成数据副作用。
• 上下文膨胀：并发结果会快速增加上下文长度。应对大体积结果做摘要或分页，只回填模型需要的字段。

三、中途引导：把修正权交还给用户

3.1 两种插话方式的区别

用户在任务执行中插话，常见有两种实现层次。

第一种在应用编排层实现。应用缓存用户的新消息，等当前轮次结束后再发起新请求，或者在下一个检查点把新消息拼接到上下文中。这种方式实现简单，但修正有延迟，而且模型已经生成的内容往往需要整体丢弃或重算。

第二种在服务接口层实现。服务端允许客户端在生成过程中提交修改指令，并保证已完成的步骤、已产生的工具结果保留下来，只让后续生成围绕新要求展开。这种方式的延迟更低，已完成的工作也不浪费，但对服务端的状态管理提出了更高要求。

两者的差别可以用一句话概括：前者是”停下来重新开始”，后者是”在原有进度上转向”。

3.2 引导的语义与状态设计

无论在哪一层实现，引导都要回答三个问题。

• 哪些内容已经确定：已经完成的文件修改、已经返回的工具结果，应当保留。
• 哪些内容需要作废：尚未执行的工具调用、基于旧要求生成的中间推理，应当取消或标记为过期。
• 新要求如何合并：新指令与原任务目标冲突时，应以用户最新表述为准，并在必要时向用户确认。

工程上常用版本号标记每一次引导。每次修改指令都带上递增的版本号，旧版本产生的结果在写入前检查版本，过期则丢弃。这样可以避免竞态条件导致旧结果覆盖新要求。

下面的示例展示应用侧如何用队列接收引导指令，并在检查点统一消费。

import asyncio
class Session:
    def __init__(self):
        self.history: list[dict] = []   # 已完成的步骤与结果
        self.steer_queue: asyncio.Queue[str] = asyncio.Queue()
        self.version = 0
    async def steer(self, instruction: str) -> None:
        # 用户随时可提交修改指令
        await self.steer_queue.put(instruction)
    def drain_steers(self) -> list[str]:
        # 在每个检查点统一读取，并递增版本号
        items = []
        while not self.steer_queue.empty():
            items.append(self.steer_queue.get_nowait())
        if items:
            self.version += 1
        return items
    def is_stale(self, result_version: int) -> bool:
        # 旧版本产生的结果不再写入
        return result_version < self.version

3.3 设计建议

• 检查点要足够密集：每完成一个工具调用或一个文件修改，就设置一个检查点，减少引导的响应延迟。
• 给用户反馈：引导提交后，界面应显示”已接收，将在下一步生效”之类的状态，避免用户重复提交。
• 记录引导历史：修改记录有助于回溯，也能帮助模型理解用户的意图变化。

四、会话中途切换推理强度与缓存保留

4.1 推理强度的含义

推理强度（thinking level）通常指模型在生成前分配的内部计算预算。预算越高，模型越可能进行多步推导、自检和方案比较，延迟和成本也随之上升。简单任务，例如列出目录结构、重命名变量，使用低强度即可；复杂任务，例如跨模块的并发缺陷分析、架构方案权衡，则值得使用高强度。

固定使用单一强度是一种常见的浪费。更合理的做法是按阶段切换：探索阶段用低强度快速扫描，设计阶段切换到高强度，执行阶段再回到中等强度。

4.2 缓存为什么重要

大模型服务通常对请求的公共前缀做缓存。若新请求的前缀与之前某次请求一致，服务端可以复用已计算的中间结果，降低延迟与费用。长会话中，系统指令、工具定义和历史消息构成了最主要的公共前缀，缓存价值很高。

问题在于，如果每次切换推理强度都导致前缀改变，缓存就会失效，切换的收益会被重新计算的成本抵消。

4.3 工程策略

能否在切换强度后保留缓存，取决于具体服务的实现，开发者应通过实测确认命中情况。在设计上，可以遵循以下原则：

• 稳定内容放前面：系统指令、工具定义、长期背景资料保持不变，并放在请求的最前部。
• 可变参数放后面：推理强度、临时指令等随任务变化的字段放在请求尾部，减少对前缀的影响。
• 监控命中率：记录每次请求的缓存命中情况，切换策略上线前后对比，用数据判断是否值得频繁切换。

示意性的请求结构如下。其中字段名仅用于说明思路，实际字段请参照所用服务的文档。

STABLE_SYSTEM_PROMPT = "你是一名资深后端工程师，回答需给出依据与风险。"
def build_request(history: list[dict], user_msg: str, level: str) -> dict:
    return {
        "system": STABLE_SYSTEM_PROMPT,   # 稳定前缀，利于缓存复用
        "messages": history + [{"role": "user", "content": user_msg}],
        "thinking_level": level,          # 可变参数放在末尾
    }
# 探索阶段低强度，设计阶段高强度
req_explore = build_request([], "列出 src 目录的模块划分", "low")
req_design = build_request([], "评估两种并发模型的取舍", "high")

五、提示词设计：防偏移与指令遵循

长任务中，模型常见两类问题。其一是偏移，即执行到一半逐渐忘记原始目标，转向次要细节。其二是指令遵循不稳定，同一条约束时而生效，时而被忽略。

5.1 结构化的任务说明

提示词应明确目标、范围、约束和验收标准。模糊的描述容易让模型自行补全，补全的内容往往偏离实际需求。

一个可复用的骨架如下：

• 目标：一句话说明要达成的结果。
• 范围：允许修改的文件或模块，以及明确排除的部分。
• 约束：必须遵守的规则，按优先级排列。
• 验收：可以被检查的完成条件，例如测试通过、接口字段齐全。

5.2 约束分级与周期复述

把约束分为”硬性约束”和”偏好”两级。硬性约束违反即视为失败，例如不得修改公共接口；偏好允许在权衡后让步，例如优先使用标准库。在长对话中，可以每隔若干轮把硬性约束重新放入上下文，对抗注意力衰减。

5.3 用测试集验证指令遵循

提示词修改后，应使用一组固定的测试样例验证效果。样例包括正常输入、边界输入和刻意的冲突输入。通过率下降时，回退修改，而不是凭主观感觉判断。

六、个性化与写作风格：对抗 AI 腔

大量生成文本带有明显的模板感：开头客套，段落之间堆砌排比，结尾重复总结，观点空泛。读者对这类文字的疲劳感正在上升，直接影响产品体验与内容可信度。

6.1 问题的来源

这类风格多半来自两个原因。一是训练数据与奖励信号偏好”看起来完整”的表达，模型倾向于用固定句式铺陈。二是提示词对受众和语气缺少约束，模型只能采用最安全的通用写法。

6.2 可落地的控制方法

• 明确受众：写清读者是谁、已有哪些背景知识、阅读目的是什么。
• 给出风格样例：提供两到三段符合目标风格的范文，比抽象描述有效。
• 设置禁用项：列出不希望出现的句式和套话，例如无意义的开场白、过度的结论升华。
• 负面示例：展示一段典型的模板化文字并标注问题，帮助模型识别需要避免的模式。
• 后处理检查：用简单规则扫描高频套话与重复句式，命中则要求模型重写该段。

6.3 个性化的实现方式

个性化不等于堆砌用户资料。更有效的做法是把用户的偏好转化为可执行的规则，例如”偏好先给结论再给依据””代码示例优先使用标准库”。规则应可被检验，并允许用户随时修改，避免模型基于不完整的历史推断用户意图。

七、落地检查清单

在把上述能力接入实际系统前，可以逐项核对：

1. 工具调用是否带唯一标识，结果回填是否依赖标识而非位置。
2. 每个工具调用是否设置超时，超时后模型是否得到明确的缺失提示。
3. 引导指令是否有检查点与版本号，过期结果是否被丢弃。
4. 推理强度切换是否经过实测，缓存命中率是否被监控。
5. 稳定的系统指令是否位于请求前部，可变参数是否位于尾部。
6. 提示词是否有固定测试集，修改后是否做过回归验证。
7. 写作类输出是否有风格样例、禁用项和后处理检查。

这三类能力的共同点是把”等待”和”打断”变成可设计的状态转换。把工具调用、用户输入和推理预算都纳入明确的状态管理，系统在长任务中会更可预测，也更容易排查问题。