Agent Harness技术解析：如何高效切换大语言模型实现业务解耦

1次阅读

没有评论

共计 3150 个字符，预计需要花费 8 分钟才能阅读完成。

在实际业务中调用不同厂商的大语言模型（LLM）时，开发者常遇到以下典型问题：

接口规范不统一 ：OpenAI 使用/v1/chat/completions，而 Anthropic 的 Claude 采用/v1/messages 作为对话接口
认证方式差异：OpenAI 的 API Key 放在 Authorization 头，部分厂商要求 X -API-Key 的自定义头
响应结构异构：同样的 ” 你好 ” 请求，不同模型返回的 JSON 字段结构可能完全不同
计费单元混乱：GPT 系列按 token 计费，Claude 按字符数计算，需要额外转换

这些差异导致业务代码与具体模型实现强耦合，每次切换模型都需要大量修改代码。更糟糕的是，当需要做 A / B 测试或多模型降级时，这种耦合会让系统变得难以维护。

Agent Harness 采用经典的设计模式组合：

适配器模式：为每个 LLM 提供商实现统一的接口适配器
工厂模式：通过统一入口动态创建具体模型实例

这种设计使得业务层只需要依赖抽象的 LLM 接口，具体实现可以在运行时决定。

我们定义四个核心接口规范：

输入规范：

class ChatInput(BaseModel):
    messages: List[Dict[str, str]]  # 对话历史
    temperature: float = 0.7
    max_tokens: int = 1024

输出规范：

class ChatOutput(BaseModel):
    content: str             # 响应内容
    input_tokens: int        # 输入 token 数
    output_tokens: int       # 输出 token 数
    latency: float           # 请求耗时(秒)

错误处理：

class LLMError(Exception):
    def __init__(self, provider: str, code: int, message: str):
        self.provider = provider
        self.code = code
        self.message = message

通过 YAML 配置文件实现模型热切换：

models:
  gpt-4:
    provider: openai
    api_key: ${OPENAI_KEY}
    endpoint: https://api.openai.com/v1
  claude-2:
    provider: anthropic
    api_key: ${ANTHROPIC_KEY}
    endpoint: https://api.anthropic.com/v1

系统启动时会监控配置文件变化，自动重新加载模型实例，无需重启服务。

from abc import ABC, abstractmethod
from typing import Dict, List

class BaseLLM(ABC):
    @abstractmethod
    async def chat(self, input: ChatInput) -> ChatOutput:
        """核心对话方法"""
        pass

    @abstractmethod
    def calculate_tokens(self, text: str) -> int:
        """计算文本 token 数"""
        pass

import openai
from .base import BaseLLM

class OpenAILlm(BaseLLM):
    def __init__(self, api_key: str, endpoint: str):
        openai.api_key = api_key
        openai.api_base = endpoint

    async def chat(self, input: ChatInput) -> ChatOutput:
        try:
            start = time.time()
            resp = await openai.ChatCompletion.acreate(
                model="gpt-4",
                messages=input.messages,
                temperature=input.temperature,
                max_tokens=input.max_tokens
            )
            return ChatOutput(content=resp.choices[0].message.content,
                input_tokens=resp.usage.prompt_tokens,
                output_tokens=resp.usage.completion_tokens,
                latency=time.time() - start)
        except Exception as e:
            raise LLMError("openai", 500, str(e))

from importlib import import_module

class LlmFactory:
    _providers = {
        'openai': 'adapters.openai.OpenAILlm',
        'anthropic': 'adapters.anthropic.ClaudeLlm'
    }

    @classmethod
    def create(cls, provider: str, **kwargs) -> BaseLLM:
        if provider not in cls._providers:
            raise ValueError(f"Unsupported provider: {provider}")

        module_path, class_name = cls._providers[provider].rsplit('.', 1)
        module = import_module(module_path)
        llm_class = getattr(module, class_name)
        return llm_class(**kwargs)

我们对三种典型场景进行了基准测试（测试环境：AWS t3.xlarge）：

首次加载：初始化模型适配器 + 建立连接池
OpenAI: 1200ms ± 200ms
Claude: 800ms ± 150ms
热切换：从 GPT- 4 切换到 Claude-2
平均耗时: 300ms (主要来自证书验证)
并发请求：100 个并行对话请求
内存增长: ≤50MB/ 模型实例

采用以下方法控制内存占用：

使用 __slots__ 减少适配器实例内存开销
实现 LRU 缓存淘汰机制，自动卸载 30 分钟未使用的模型
限制并行加载的模型数量（默认最多 5 个）

对于高频访问场景，建议调整默认参数：

import httpx

async with httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,
        max_keepalive_connections=20,
        keepalive_expiry=300
    )
) as client:
    # 传递给适配器

主要模型的 token 计算规则：