构建模型无关的 AI Agent

大多数 AI Agent 框架在诞生之初就绑定了特定的模型供应商。LangChain 最初围绕 OpenAI 构建，Claude Code 自然是 Anthropic 专属。但实际使用中，你经常需要在不同模型之间切换——因为成本、延迟、能力匹配，或者单纯不想被锁定。

本文展示如何构建一个能与任何模型配合工作的 Agent——Claude、GPT、DeepSeek、Llama 或本地部署的开源模型——只需修改一行配置。

为什么模型无关很重要

什么是模型无关架构

模型无关（Model-Agnostic）意味着你的 Agent 核心逻辑不依赖任何特定模型的 API 格式。Agent 循环——观察 → 思考 → 行动 → 观察——无论哪个模型驱动「思考」步骤，都完全一样。

架构分为三层：

1. Agent 核心——ReAct 循环、工具执行、记忆管理（与模型无关）
2. 适配器层——将核心的统一格式转换为每个模型的具体 API
3. 模型提供商——Claude、GPT、DeepSeek、Llama 等（可替换）

┌─────────────────────────────────┐
│         Agent 核心循环            │  ← 永不改变
│  观察 → 思考 → 行动 → 观察       │
└──────────────┬──────────────────┘
               │ 统一接口
┌──────────────▼──────────────────┐
│          适配器层                 │  ← 按模型切换
│  ┌────────┐ ┌──────┐ ┌───────┐  │
│  │Claude  │ │ GPT  │ │DeepSk │  │
│  │适配器  │ │适配器│ │适配器 │  │
│  └───┬────┘ └──┬───┘ └───┬───┘  │
└──────┼─────────┼─────────┼──────┘
       │         │         │
   Anthropic  OpenAI  DeepSeek API

适配器接口

每个适配器实现相同的接口。以下是契约：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any

@dataclass
class AgentResponse:
    """统一响应格式，来自任何模型。"""
    content: str | None        # 最终文本答案（is_final=True 时）
    tool_call: dict | None     # 工具调用请求（is_final=False 时）
    is_final: bool             # True=完成, False=需要调用工具
    usage: dict                # Token 用量: {"input": N, "output": M}

class ModelAdapter(ABC):
    """每个模型适配器必须实现此接口。"""

    @abstractmethod
    def chat(self, messages: list[dict],
             tools: list[dict] | None = None,
             temperature: float = 0.7,
             max_tokens: int = 1000) -> AgentResponse:
        """发送消息 + 工具 → 接收响应或工具调用。"""
        ...

    @abstractmethod
    def normalize_tools(self, tools: list[dict]) -> list[dict]:
        """将统一工具 schema 转换为模型特定格式。"""
        ...

    @abstractmethod
    def normalize_messages(self, messages: list[dict]) -> list[dict]:
        """将统一消息格式转换为模型特定格式。"""
        ...

实现真正的适配器

以下是三种最常见模型家族的具体实现。注意每种模型处理工具调用的方式有何不同。

OpenAI 适配器（GPT-4o, GPT-4, GPT-3.5）

from openai import OpenAI

class OpenAIAdapter(ModelAdapter):
    def __init__(self, model="gpt-4o", api_key=None, base_url=None):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.model = model

    def normalize_tools(self, tools):
        # OpenAI 使用标准 function-calling 格式——改动最小
        return [{"type": "function", "function": t} for t in tools]

    def normalize_messages(self, messages):
        # OpenAI 格式是基线；工具结果使用 role "tool"
        return messages  # 已经是正确格式

    def chat(self, messages, tools=None, temperature=0.7, max_tokens=1000):
        kwargs = dict(
            model=self.model,
            messages=self.normalize_messages(messages),
            temperature=temperature,
            max_tokens=max_tokens
        )
        if tools:
            kwargs["tools"] = self.normalize_tools(tools)

        resp = self.client.chat.completions.create(**kwargs)
        msg = resp.choices[0].message

        return AgentResponse(
            content=msg.content,
            tool_call={
                "name": msg.tool_calls[0].function.name,
                "arguments": msg.tool_calls[0].function.arguments
            } if msg.tool_calls else None,
            is_final=msg.tool_calls is None,
            usage={
                "input": resp.usage.prompt_tokens,
                "output": resp.usage.completion_tokens
            }
        )

Anthropic 适配器（Claude Sonnet, Claude Opus）

import anthropic

class AnthropicAdapter(ModelAdapter):
    def __init__(self, model="claude-sonnet-4-20250514", api_key=None):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = model

    def normalize_tools(self, tools):
        # Anthropic 使用不同的工具格式——没有 "type":"function" 包装
        normalized = []
        for tool in tools:
            inner = tool.get("function", tool)  # 如果嵌套则解包
            normalized.append({
                "name": inner["name"],
                "description": inner.get("description", ""),
                "input_schema": inner.get("parameters",
                    {"type": "object", "properties": {}})
            })
        return normalized

    def normalize_messages(self, messages):
        # Anthropic 需要将 system prompt 提取为单独参数
        normalized = []
        for msg in messages:
            if msg["role"] == "system":
                continue  # 单独处理
            if msg["role"] == "tool":
                # Anthropic 在 user 消息中使用 "tool_result" 块
                normalized.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": msg.get("tool_call_id", "unknown"),
                        "content": msg["content"]
                    }]
                })
            else:
                normalized.append({"role": msg["role"],
                                   "content": msg["content"]})
        return normalized

    def chat(self, messages, tools=None, temperature=0.7, max_tokens=1000):
        system = next((m["content"] for m in messages
                       if m["role"] == "system"), None)
        normalized_msgs = self.normalize_messages(messages)

        kwargs = dict(
            model=self.model,
            messages=normalized_msgs,
            max_tokens=max_tokens,
            temperature=temperature
        )
        if system:
            kwargs["system"] = system
        if tools:
            kwargs["tools"] = self.normalize_tools(tools)

        resp = self.client.messages.create(**kwargs)

        # 从响应中提取 tool use 块
        tool_calls = [
            block for block in resp.content
            if block.type == "tool_use"
        ]

        return AgentResponse(
            content=resp.content[0].text if resp.content[0].type == "text"
                    else None,
            tool_call={
                "name": tool_calls[0].name,
                "arguments": tool_calls[0].input,
                "id": tool_calls[0].id
            } if tool_calls else None,
            is_final=len(tool_calls) == 0,
            usage={
                "input": resp.usage.input_tokens,
                "output": resp.usage.output_tokens
            }
        )

OpenAI 兼容适配器（DeepSeek, vLLM, Ollama, 本地模型）

许多模型（DeepSeek、通过 vLLM/Ollama 运行的 Llama、Groq）使用 OpenAI 兼容 API。一个适配器覆盖所有——只需修改 base_url：

# DeepSeek——简单任务便宜 10 倍，中文表现优秀
agent = ModelAgnosticAgent(
    OpenAIAdapter(
        model="deepseek-chat",
        api_key="sk-xxx",
        base_url="https://api.deepseek.com/v1"
    ),
    tools, prompt
)

# 本地 Llama 3 via Ollama——零成本，完全隐私
agent = ModelAgnosticAgent(
    OpenAIAdapter(
        model="llama3:70b",
        api_key="ollama",  # Ollama 忽略 key
        base_url="http://localhost:11434/v1"
    ),
    tools, prompt
)

# Groq——最快推理速度，适合实时场景
agent = ModelAgnosticAgent(
    OpenAIAdapter(
        model="llama-3.1-70b-versatile",
        api_key="gsk_xxx",
        base_url="https://api.groq.com/openai/v1"
    ),
    tools, prompt
)

工具格式归一化

不同提供商的工具 schema 略有差异。核心洞察：以 OpenAI 的 function-calling 格式作为内部标准，让每个适配器转换为其原生格式。

模型选择策略

拥有模型无关架构后，你可以根据特征将任务路由到最优模型：

def smart_route(task: str) -> ModelAdapter:
    """根据启发式规则将任务路由到最佳模型。"""
    if any(kw in task.lower() for kw in ["code", "debug", "math", "logic"]):
        return AnthropicAdapter(model="claude-opus-4-20250514")
    if any(kw in task.lower() for kw in ["中文", "chinese", "翻译"]):
        return OpenAIAdapter(model="deepseek-chat",
                            base_url="https://api.deepseek.com/v1")
    if len(task) < 100:  # 简单短查询
        return OpenAIAdapter(model="gpt-4o-mini")
    return AnthropicAdapter(model="claude-sonnet-4-20250514")  # 默认

完整 Agent 实现

以下是完整的模型无关 Agent。核心循环永不改变——只有适配器会变：

class ModelAgnosticAgent:
    def __init__(self, model: ModelAdapter, tools: list[dict],
                 system_prompt: str):
        self.model = model
        self.tools = tools
        self.messages = [{"role": "system", "content": system_prompt}]
        self.total_cost = 0.0

    def run(self, user_input: str, max_turns: int = 20) -> str:
        self.messages.append({"role": "user", "content": user_input})
        turns = 0

        while turns < max_turns:
            response = self.model.chat(
                self.messages, self.tools,
                temperature=0.7 if turns == 0 else 0.4  # 逐步降温
            )
            turns += 1

            if response.is_final:
                return response.content

            # 执行工具并将结果反馈
            tool_name = response.tool_call["name"]
            tool_args = response.tool_call.get("arguments", {})
            result = self._execute_tool(tool_name, tool_args)

            self.messages.append({
                "role": "assistant",
                "content": None,
                "tool_calls": [{
                    "id": f"call_{turns}",
                    "type": "function",
                    "function": {
                        "name": tool_name,
                        "arguments": json.dumps(tool_args)
                            if isinstance(tool_args, dict)
                            else tool_args
                    }
                }]
            })
            self.messages.append({
                "role": "tool",
                "tool_call_id": f"call_{turns}",
                "content": json.dumps(result)
            })

        return "已达最大轮次，任务未完成。"

测试模型无关层

如何知道适配器工作正常？用同一个工具调用任务跨所有模型测试：

def test_adapter(adapter: ModelAdapter):
    """验证适配器是否正确处理工具调用。"""
    tools = [{
        "name": "get_weather",
        "description": "获取指定城市的当前天气。",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"}
            },
            "required": ["city"]
        }
    }]

    messages = [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": "东京今天天气怎么样？"}
    ]

    response = adapter.chat(messages, tools)

    assert not response.is_final, "应该请求工具调用"
    assert response.tool_call["name"] == "get_weather", \
        f"调了错误的工具: {response.tool_call['name']}"
    assert "东京" in str(response.tool_call.get("arguments", "")), \
        "缺少城市参数"

# 对所有适配器运行测试
for name, adapter in [
    ("OpenAI", OpenAIAdapter()),
    ("Anthropic", AnthropicAdapter()),
    ("DeepSeek", OpenAIAdapter(base_url="https://api.deepseek.com/v1"))
]:
    try:
        test_adapter(adapter)
        print(f"✅ {name}: 通过")
    except Exception as e:
        print(f"❌ {name}: 失败 — {e}")

现有框架对比（以及何时自己造轮子）

核心要点

1. 定义统一的适配器接口——一个 chat() 方法，一个 normalize_tools()，一个 normalize_messages()。其他都是实现细节。
2. 内部统一使用 OpenAI 工具格式——这是事实标准，大多数提供商都实现或可以转换为此格式。
3. 用同一个工具调用场景测试每个适配器——工具格式转换 bug 是沉默的，在生产环境难以调试。
4. 智能路由省钱——简单查询用便宜模型（DeepSeek 约 $0.14/百万 token），复杂推理用高级模型（Claude 约 $3/百万 token）。成本差距可达 20 倍。
5. 不要过度设计——如果你只用一个模型，跳过这一切。但如果你在构建能持久的东西，适配器模式是防止厂商锁定的廉价保险。