不改 Agent 逻辑,先用四个请求级杠杆压低运行成本

2026-08-27 27 预计阅读时间: 1 分钟
来源: azure.microsoft.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

Agent 成本优化不一定要从重写工作流、删减工具或修改推理步骤开始。来源摘要指出,Microsoft Foundry 提供了四个作用于每次请求的杠杆,而且这些优化可以发生在 Agent 逻辑执行之前。这意味着团队可以先在统一请求入口控制成本,再决定是否改动业务代码。

摘要没有列出四个杠杆的具体名称。下面不把某组产品配置冒充为原文结论,而是给出一种可以落地的工程映射:模型路由、输入预算、输出上限和结果缓存。它们共同指向一个核心思路:把成本控制放在 Agent 与模型之间的请求层。

先算清每次请求为什么贵

一次 Agent 调用的成本不只是最终回答产生的 token。完整链路通常包含系统提示词、对话历史、检索结果、工具返回值、模型输出,以及 Agent 在一次任务中发起的多轮模型调用。

可以用下面的简化公式建立成本基线:

单次任务成本 ≈ Σ(输入 token × 输入单价 + 输出 token × 输出单价)
                + 工具与基础设施成本

这里的求和很重要。一个任务如果调用模型五次,即使每次只节省 10%,累计收益也可能高于只压缩最终回答。

上线优化前,至少记录这些字段:

  • 使用的模型或部署名称
  • 输入、输出和缓存命中的 token 数
  • 一个用户任务触发的模型调用次数
  • 请求延迟、错误率与重试次数
  • 任务成功率或人工验收结果
  • 每个任务、用户和业务场景的估算成本

不要只看平均成本。P50 可以描述常规流量,但真正推高账单的往往是携带超长历史、反复调用工具或持续重试的 P95 请求。

四个请求级杠杆如何工作

1. 按任务难度路由模型

分类、字段提取、格式转换等确定性较高的请求,可以优先交给成本较低的模型;复杂规划、长文综合和高风险判断再升级到能力更强的模型。

路由条件应当可审计,例如任务类型、输入长度、是否涉及敏感操作和低成本模型的置信度。仅凭提示词中的某个关键词切换模型,容易被用户输入误触发,甚至形成提示注入入口。

2. 给输入上下文设预算

Agent 很容易把全部聊天记录、搜索结果和工具日志原样发送给模型。更稳妥的做法是为不同上下文分配预算:系统指令保留多少、最近消息保留多少、检索片段最多多少、工具输出最多多少。

截断时不能只按字符数从尾部切割。应该优先保留安全规则、当前任务、关键事实和工具错误,再压缩较旧的历史信息。对长会话,可以维护经过验证的滚动摘要。

3. 限制输出,并让格式更确定

设置输出 token 上限能直接约束最坏情况成本。对于分类、路由和结构化抽取任务,还可以要求模型返回紧凑 JSON,避免生成解释性段落。

上限不是越小越好。截断 JSON、代码或工具参数会触发解析失败与重试,最终反而更贵。应同时监控完成原因、解析失败率和重试次数。

4. 缓存可复用结果

相同或等价的请求没有必要反复推理。适合缓存的场景包括稳定文档问答、固定分类规则、重复摘要和不依赖实时数据的格式转换。

缓存键不能只使用用户文本。模型版本、系统提示词版本、知识库版本、租户和权限范围都可能影响结果。涉及个人数据、实时价格或动态授权时,要缩短 TTL,或者直接绕过缓存。

可以这样实践:在 Agent 前增加成本控制网关

下面是一个可运行、可改造的 Python 示例。它使用标准库实现四项控制:按输入复杂度选择部署、限制上下文长度、设置输出上限,并对低风险请求做本地缓存。

这是工程示例,不代表 Microsoft Foundry 的具体接口名称。运行前需要把 FOUNDRY_ENDPOINTFOUNDRY_API_KEY 以及两个部署名改成你的实际配置;端点需要提供 OpenAI 兼容的 Chat Completions 接口。

import hashlib
import json
import os
import time
import urllib.request

ENDPOINT = os.environ['FOUNDRY_ENDPOINT'].rstrip('/')
API_KEY = os.environ['FOUNDRY_API_KEY']
CHEAP_MODEL = os.getenv('CHEAP_MODEL', 'small-model-deployment')
STRONG_MODEL = os.getenv('STRONG_MODEL', 'strong-model-deployment')
CACHE_TTL_SECONDS = 300
MAX_CONTEXT_CHARS = 12000

cache = {}


def choose_model(task, text):
    complex_tasks = {'planning', 'code_review', 'risk_analysis'}
    if task in complex_tasks or len(text) > 6000:
        return STRONG_MODEL
    return CHEAP_MODEL


def trim_context(text):
    if len(text) <= MAX_CONTEXT_CHARS:
        return text
    return text[-MAX_CONTEXT_CHARS:]


def cache_key(model, task, text, prompt_version='v1'):
    raw = json.dumps(
        [model, task, text, prompt_version],
        ensure_ascii=False,
        separators=(',', ':'),
    )
    return hashlib.sha256(raw.encode('utf-8')).hexdigest()


def call_agent(task, text, cacheable=True):
    text = trim_context(text)
    model = choose_model(task, text)
    key = cache_key(model, task, text)
    now = time.time()

    if cacheable and key in cache:
        expires_at, result = cache[key]
        if expires_at > now:
            return {'cached': True, 'model': model, 'content': result}

    payload = {
        'model': model,
        'messages': [
            {
                'role': 'system',
                'content': 'Answer accurately and concisely. Return only the requested result.',
            },
            {'role': 'user', 'content': text},
        ],
        'max_tokens': 400,
        'temperature': 0,
    }

    request = urllib.request.Request(
        f'{ENDPOINT}/v1/chat/completions',
        data=json.dumps(payload).encode('utf-8'),
        headers={
            'Authorization': f'Bearer {API_KEY}',
            'Content-Type': 'application/json',
        },
        method='POST',
    )

    with urllib.request.urlopen(request, timeout=60) as response:
        body = json.loads(response.read().decode('utf-8'))

    result = body['choices'][0]['message']['content']
    if cacheable:
        cache[key] = (now + CACHE_TTL_SECONDS, result)

    usage = body.get('usage', {})
    return {
        'cached': False,
        'model': model,
        'content': result,
        'input_tokens': usage.get('prompt_tokens'),
        'output_tokens': usage.get('completion_tokens'),
    }


if __name__ == '__main__':
    answer = call_agent(
        task='classification',
        text='Classify this ticket as billing, technical, or account: I was charged twice.',
    )
    print(json.dumps(answer, ensure_ascii=False, indent=2))

可以这样运行:

export FOUNDRY_ENDPOINT='https://your-endpoint.example.com'
export FOUNDRY_API_KEY='replace-with-your-key'
export CHEAP_MODEL='your-small-deployment'
export STRONG_MODEL='your-strong-deployment'
python agent_gateway.py

生产环境中应把进程内缓存替换为带租户隔离的 Redis 等共享存储,并增加请求超时、退避重试、并发限制、审计日志和敏感信息处理。API 密钥也应由密钥管理服务提供,不应写进代码仓库。

用实验验证,而不是只看账单下降

每个杠杆都可能牺牲质量。更换模型可能降低复杂任务成功率,压缩上下文可能遗漏早期约束,限制输出可能造成截断,缓存则可能返回过期结果。因此,成本实验必须同时设置质量护栏。

建议以同一批代表性任务进行 A/B 测试,并比较:

  • 每个成功任务的成本,而不是单次请求成本
  • 任务成功率和人工评分
  • P50、P95 延迟
  • 平均模型调用次数
  • 解析失败与自动重试率
  • 缓存命中率和过期答案率

推荐采用渐进顺序:先补齐观测数据,再设置输出和上下文上限,然后引入模型路由,最后缓存明确可复用的请求。每次只调整一个变量,保留回滚开关,并按任务类型分别评估。

真正有效的 Agent 成本优化,不是让每个请求都使用最便宜的模型,而是让每个任务以最低的可接受成本完成。把四类控制放到统一请求层后,团队可以在不侵入 Agent 业务逻辑的情况下快速试验,同时保留质量、安全和审计边界。


相关推荐