智能体大模型全路径：预算在四个环节之间的分配

原创

IT大佬 jzit-top

发布于 2026-11-09 18:27:23

140

一次智能体请求从进入系统到返回用户，要穿过路由、上下文组装、模型推理、工具执行四个环节。大多数实现只对模型调用这一个环节设置了预算——max_tokens、temperature、超时时间——其余三个环节是开放式的。结果是：路由把简单问题送给了最大的模型，上下文把所有可用内容塞满，工具在串行等待中耗尽延迟预算。全路径的工程问题，本质上是一个预算分配问题：token、延迟、成本、信任，这四个预算在环节之间此消彼长，任何一个环节独占预算，其余环节就会在无预算的状态下失控。

环节一：入口路由——不是每个请求都值得一次大模型调用

最昂贵的浪费发生在请求进入系统的第一秒：一个"今天天气怎么样"被送进了 70B 模型，配上 8000 token 的 RAG 上下文。路由层的职责是把请求分派到与复杂度匹配的处理路径，而不是让主模型自我评估"我需不需要深度思考"。

路由应该基于可观测的请求特征做快速分类，而不是再调一次模型：

def route(query: str, session: dict) -> dict:
    n = len(query)
    has_code = "```" in query
    has_multi_question = query.count("？") + query.count("?") > 2
    is_followup = bool(session.get("last_topic"))
    needs_tools = any(k in query for k in ("查询", "下单", "退款", "对比"))

    if has_code or has_multi_question:
        return {"path": "deep", "model": "large", "max_steps": 8}
    if needs_tools or is_followup:
        return {"path": "agent", "model": "mid", "max_steps": 4}
    if n < 30 and not has_code:
        return {"path": "fast", "model": "small", "max_steps": 1}
    return {"path": "standard", "model": "mid", "max_steps": 3}

max_steps 随路径一起下发，而不是在循环内部硬编码，这意味着路由层同时决定了"用哪个模型"和"允许走多远"。一个快速路径请求根本不应该有工具循环的机会——不是靠执行时拒绝，而是在入口处就不给它这个预算。

环节二：上下文组装——按优先级分配，而不是按相似度截断

检索回来 50 个块、历史对话 30 轮、系统提示 2000 字，模型上下文窗口是 8000。最常见的做法是"取 top-k、按时间截断"，这个做法把窗口当成一个无差别的容器。实际上窗口里的每一段内容价值差异极大：系统约束一旦被截断，模型行为立刻偏离；最相关的两个块被截掉，回答直接失去依据。

正确的组装是显式的预算分配：先给系统约束和当前问题预留硬额度，剩余空间按优先级填充，任何一段被裁剪时都记录裁剪日志：

def assemble(system: str, query: str, chunks: list, history: list,
             window: int = 8000) -> dict:
    reserved = len(system) + len(query) + 200   # 200 留给输出分隔
    remaining = window - reserved
    if remaining <= 0:
        return {"error": "prompt_exceeds_window", "over_by": -remaining}

    parts, used, dropped = [], 0, []
    for c in sorted(chunks, key=lambda x: x["score"], reverse=True):
        if used + len(c["text"]) <= remaining * 0.6:   # 检索块占 60%
            parts.append(c["text"]); used += len(c["text"])
        else:
            dropped.append({"id": c["id"], "reason": "chunk_budget"})

    h_budget = remaining - used
    for turn in reversed(history):
        if len(turn) <= h_budget:
            parts.insert(0, turn); h_budget -= len(turn)
        else:
            dropped.append({"turn": turn[:20], "reason": "history_budget"})
            break

    return {"prompt": system + "\n\n" + "\n".join(parts) + "\n\n" + query,
            "dropped": dropped, "used": used}

dropped 列表是这段代码的工程核心。被裁剪的块和对话轮次应该进入观测日志，因为"模型答得不好"的很大一部分原因可以从这里直接看出来：关键的第三块因为检索分数排在第五位，被 60% 的检索预算挤掉了。

环节三：模型推理——流式中断的重试语义

流式返回是改善首字延迟的标准手段，但它和重试机制之间有一个安静的冲突。非流式调用失败时，重试是干净的：一个请求失败、重新发一个、得到完整响应。流式调用失败时，前 300 个 token 已经推给了用户，重试会从头再来一遍，用户看到两段拼在一起的内容。

工程上的做法是把"是否已经向用户输出"作为重试决策的一部分：

class StreamGuard:
    def __init__(self, client, max_retries=2):
        self.client, self.max_retries = client, max_retries
        self.emitted = False

    async def run(self, prompt, on_token):
        for attempt in range(self.max_retries + 1):
            try:
                async for tok in self.client.stream(prompt):
                    if not self.emitted:
                        self.emitted = True
                    on_token(tok)
                return {"ok": True, "attempts": attempt + 1}
            except Exception as e:
                if self.emitted:
                    return {"ok": False, "error": "stream_broken_after_emit",
                            "detail": str(e), "attempts": attempt + 1}
                continue
        return {"ok": False, "error": "retries_exhausted"}

emitted 一旦为真，任何异常都直接返回，不再重试。这个选择的代价是用户拿到一个不完整的回答，收益是不会出现重复或矛盾的内容。要同时改善两者，需要在上层做"可断点续流"的协议——那是一个更大的设计决策，但它的前提正是这个 guard 先区分了"失败发生在输出之前还是之后"。

环节四：工具执行——并行化之前先做依赖分析

Agent 调工具最常见的低效是串行：模型一轮返回三个独立的查询，执行器逐个等。并行化能显著压缩延迟，但不能无脑并行——工具之间可能有隐式依赖，比如"先查订单，再查物流"需要前者的返回值。

一个够用的做法是按模型给出的依赖声明分组，组内并行、组间串行：

import asyncio

async def run_tool_plan(calls: list, executor) -> list:
    # calls: [{"id": 1, "tool": "x", "args": {}, "depends_on": []}]
    done, results = set(), {}
    while len(done) < len(calls):
        ready = [c for c in calls if c["id"] not in done
                 and all(d in done for d in c["depends_on"])]
        if not ready:
            return [{"error": "cyclic_dependency",
                     "blocked": [c["id"] for c in calls if c["id"] not in done]}]

        batch = await asyncio.gather(
            *(executor(c["tool"], c["args"]) for c in ready),
            return_exceptions=True,
        )
        for c, r in zip(ready, batch):
            results[c["id"]] = r
            done.add(c["id"])
    return [results[c["id"]] for c in calls]

cyclic_dependency 分支不是防御性编程的装饰。模型生成的工具计划里出现循环依赖并不罕见——它会把"用 A 的结果调 B、再用 B 的结果调 A"理解成一个合理的推理链。执行器必须能识别并返回这个状态，让上层把错误反馈给模型重新规划，而不是在循环里等到超时。

全路径的收束

四个环节的预算关系是联动的：路由多给一步工具预算，工具执行就要多给延迟预算；上下文多留检索块预算，模型输出预算就要相应压缩。把它们分开配置、各自调参，得到的永远是一个局部最优、全局错配的系统。全路径的工程纪律是在入口处一次性下发一组相互协调的预算，让每个环节在拿到自己的额度时知道别的环节也在消耗同一份资源。这比事后在某个环节里做"优化"更有效，因为一旦预算分配错了，后续任何单点优化都只是在为一个错误的约束求解。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

智能创作