Agent 工作负载生产部署最佳实践

能够规划、推理并在循环中调用工具的 Agent 系统已从研究演示进入生产流水线。要可靠地部署它们，需要解决上下文增长、延迟叠加以及级联故障模式等问题——这些问题在单轮聊天中是不会出现的。本文涵盖了在生产环境中验证有效的架构模式、安全机制和成本控制，并提供了可直接适配的代码示例。

Agent 角色的模型选择

并非每个 Agent 步骤都需要相同的算力。负责将任务路由到工作节点的 supervisor 与跨仓库编辑文件的编码 Agent，其需求截然不同。Oxlo.ai 托管的模型可以清晰地映射到这些层级。

• 推理与规划：DeepSeek R1 671B MoE 用于深度推理，或 Kimi K2.6 用于高级推理和 Agent 编码
• 长时域上下文：DeepSeek V4 Flash 支持 1M 上下文，或 GLM 5 用于长时域 Agent 任务
• 通用执行：Llama 3.3 70B 或 Qwen 3 32B 用于多语言 Agent 工作流
• 快速工具调用：Minimax M2.5 用于编码和 Agent 工具调用，或 Qwen 3 Coder 30B

在运行时将任务路由到合适的模型，以平衡延迟和能力。

import openai
import os

client = openai.OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ["OXLO_API_KEY"]
)

# Example mapping to Oxlo.ai model tiers
MODEL_MAP = {
    "long_context": "deepseek-v4-flash",  # DeepSeek V4 Flash, 1M context
    "coding": "kimi-k2.6",                # Kimi K2.6
    "default": "llama-3.3-70b"            # Llama 3.3 70B
}

def route_task(prompt, context_length):
    if context_length > 100_000:
        model = MODEL_MAP["long_context"]
    elif any(k in prompt for k in ("refactor", "debug")):
        model = MODEL_MAP["coding"]
    else:
        model = MODEL_MAP["default"]
    return client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        tools=tools,
    )

上下文预算与循环控制

生产 Agent 工作负载中最大的惊喜是上下文膨胀。每个工具响应、观察结果和中间推理步骤都会被追加到对话历史中。在按 token 计费的提供商上，每次循环迭代都会产生费用。Oxlo.ai 使用按请求计费，因此每个 API 请求一个固定费率，即使工具输出很长，成本也保持可预测。不过，你仍然应该设置限制以保持延迟和准确性。

• 硬性限制迭代次数（例如最多 10 步）
• 将超过 N 步的轮次总结为压缩的草稿区
• 截断工具输出（例如每个工具响应最多 4000 字符）

class AgentLoop:
    def __init__(self, max_steps=10, max_tool_output=4000):
        self.max_steps = max_steps
        self.max_tool_output = max_tool_output
        self.history = []

    def run(self, task):
        for step in range(self.max_steps):
            response = self.llm(step, task)
            if response.finish_reason == "stop":
                return response.content
            tool_result = self.execute_tool(response.tool_calls)
            self.history.append({
                "step": step,
                "tool_result": tool_result[:self.max_tool_output]
            })

工具 Schema 强化

函数调用是 Agent 系统的粘合剂，但 Schema 会漂移，模型会 hallucinate 参数。Oxlo.ai 支持函数调用、JSON 模式和流式输出，你可以将它们组合起来以构建严格的工具边界。

• 在代码中对工具 Schema 进行版本控制，而非在 prompt 中
• 在执行前使用 Pydantic 验证参数
• 当需要结构化输出但不需要工具调用时，使用 JSON 模式
• 为每个工具调用设置超时

from pydantic import BaseModel, ValidationError
import json

class SearchArgs(BaseModel):
    query: str
    top_k: int = 5

def safe_call_tool(name, arguments):
    if name == "search":
        try:
            args = SearchArgs(**json.loads(arguments))
        except ValidationError:
            return {"error": "Invalid schema"}
        return search(**args.dict())

故障模式与重试

Agent 的失败方式与聊天机器人不同。工具超时可能使整个工作流停滞，而错误参数可能触发破坏性操作。将 LLM 视为不可靠的调用者，并在每个外部交互处包装断路器。

• 对所有有副作用的工具使用幂等性密钥
• 在 LLM 客户端上使用指数退避，而不仅仅是工具
• 在整个 Agent 循环上设置全局超时，而非每步设置
• Oxlo.ai 在热门模型上没有冷启动，因此重试风暴不会触发增加延迟的预热惩罚

import time
from functools import wraps

def with_retries(max_attempts=3, backoff=2):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(backoff ** attempt)
        return wrapper
    return decorator

并行与顺序执行

当 Agent 需要从三个 API 收集数据时，顺序工具调用会成倍增加延迟。在没有依赖关系的地方，进行扇出。

Oxlo.ai 支持流式输出且没有冷启动，因此并行调用可以在没有隐藏预热成本的情况下扩展。使用 supervisor 来聚合结果。

import concurrent.futures

def supervisor_plan(task):
    plan = llm_generate_plan(task)
    with concurrent.futures.ThreadPoolExecutor() as pool:
        futures = {
            pool.submit(call_tool, step.tool, step.args): step
            for step in plan if step.parallel
        }
        results = {futures[f]: f.result(timeout=30) for f in futures}
    return synthesize(results)

可观测性与状态管理

你无法调试你看不见的东西。记录每个中间完成结果、工具参数和原始工具响应。因为 Oxlo.ai 按请求扁平计价，你每步的成本是确定性的，这使得无需 token 计算就能将费用归因到各个 Agent 轨迹。

将状态存储在外部（Redis、Postgres），这样崩溃不会丢失进度。在每个工具调用后设置检查点。

def checkpoint(state):
    redis.setex(f"agent:{state.run_id}", 3600, json.dumps(state.to_dict()))

def step(state):
    response = client.chat.completions.create(
        model="qwen-3-32b",  # example identifier
        messages=state.messages,
        tools=state.tools,
    )
    state.add_turn(response)
    checkpoint(state)
    return state

按请求计费对 Agent 经济性的影响

基于 token 的定价会对 Agent 最频繁表现出的行为处以惩罚：将长工具输出和推理轨迹追加到上下文中。在基于 token 的提供商中，单个 Agent 循环每步可能消耗数万个 token，成本随该增长线性扩展。

Oxlo.ai 对每个 API 请求收取一个固定费用，不计 prompt 长度。对于 Agent 工作负载，这改变了经济学逻辑。你可以将完整的文件内容、堆栈跟踪或检索块传入上下文，而不必在每轮都受到按量计费的惩罚。详见 https://oxlo.ai/pricing 了解计划详情。

这并不意味着你应该忽略上下文限制。模型上下文窗口仍然限制了准确性，延迟随序列长度增加而上升。但这意味着你的成本优化策略从 token 修剪转向请求效率：将工作批处理到更少的调用中，消除冗余循环，并为每步选择合适的模型层级。

Agent 部署本质上是一个系统工程问题。能够可靠交付的团队将 LLM 视为更大控制循环中的一个组件：有限迭代、经过验证的 Schema、外部状态和可预测的成本。Oxlo.ai 提供了基于请求的推理层、模型多样性以及与 OpenAI 兼容的 API，让你专注于循环本身，而非账单。