第十章 多模型适配 — Provider 抽象层
Hermes 支持 200+ 模型(通过 OpenRouter)、Anthropic Claude、Google Gemini、AWS Bedrock、Mistral 等。核心是一个统一的传输层抽象——所有 Provider 的响应都被归一化为 NormalizedResponse,让 Agent 循环完全不需要关心具体模型。
10.1 传输层抽象
NormalizedResponse
@dataclass
class NormalizedResponse:
content: Optional[str] # 文本回复
tool_calls: Optional[List[ToolCall]] # 工具调用列表
finish_reason: str # "stop" | "tool_calls"
reasoning: Optional[str] = None # 思考过程(扩展思维模型)
usage: Optional[Usage] = None # token 使用量
provider_data: Optional[Dict] = None # Provider 原始数据为什么需要归一化?
不同 Provider 的响应格式完全不同:
| Provider | 工具调用字段 | 思考过程字段 | 停止原因 |
|---|---|---|---|
| OpenAI | response.tool_calls | 无 | finish_reason |
| Anthropic | response.content[type=tool_use] | content[type=thinking] | stop_reason |
| Gemini | functionCall | thoughtPart | finishReason |
归一化后,Agent 循环只需要处理一种格式。
10.2 Anthropic 适配器
agent/anthropic_adapter.py 是最复杂的适配器,支持多种认证和功能:
认证方式
扩展思维(Thinking)
Anthropic 模型支持 thinking budget:
# adaptive_effort 映射
effort_map = {
"low": {"budget_tokens": 1024},
"medium": {"budget_tokens": 8192},
"high": {"budget_tokens": 32768},
}思考过程通过 reasoning 字段传递给 Agent,但不发送给用户。
Prompt Caching 集成
适配器在发送请求前注入 cache breakpoint:
# system_and_3 策略:4 个缓存点
messages_with_cache = apply_cache_control(
messages,
strategy="system_and_3", # system + 最近 3 条消息
)10.3 Gemini 适配器
agent/gemini_native_adapter.py 直接调用 Gemini REST API,绕过 OpenAI 兼容层:
Schema 转换
Gemini 的工具格式与 OpenAI 不同,需要转换:
多模态支持
Gemini 原生支持图片输入:
# 图片内容转换
def _convert_image_content(image_url: str) -> dict:
return {
"inlineData": {
"mimeType": "image/png",
"data": base64.b64encode(fetch_image(image_url)).decode()
}
}Gemini Cloud Code
agent/gemini_cloudcode_adapter.py 是 Gemini 的变体,通过 Google Cloud Code API 访问:
- 使用 Google OAuth 认证
- 支持代码补全和代码生成模式
- 与
google_code_assist.py集成
10.4 Bedrock 适配器
agent/bedrock_adapter.py 通过 AWS Bedrock 访问 Claude 模型:
# 配置
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
MODEL_ID = "anthropic.claude-3-5-sonnet-20241022-v2:0"
# 使用 boto3
client = boto3.client("bedrock-runtime", region_name=AWS_REGION)
response = client.invoke_model(
modelId=MODEL_ID,
body=json.dumps(request_body),
)支持的认证
- IAM Role(EC2/ECS 自动获取)
- AWS Access Key + Secret Key
- AWS Profile
10.5 其他 Provider
OpenAI 兼容(默认路径)
大部分 Provider 都兼容 OpenAI API 格式:
client = OpenAI(
base_url=provider_url, # 任意兼容端点
api_key=api_key,
)
response = client.chat.completions.create(
model=model_name,
messages=messages,
tools=tool_schemas,
)Nous Portal
Nous Research 的自有平台:
https://portal.nousresearch.com/v1- 提供优化的模型路由和速率限制
OpenRouter
通过 OpenRouter 访问 200+ 模型:
https://openrouter.ai/api/v1- 支持 Provider 过滤、排序、价格优化
providers_allowed/providers_ignored/providers_order参数
本地模型
通过 Ollama 或 vLLM 运行本地模型:
# Ollama
ollama serve
hermes model # 选择 ollama/llama3
# vLLM
python -m vllm.entrypoint.openai.api_server --model meta-llama/Meta-Llama-3-8B
hermes model # 选择 custom endpoint10.6 模型元数据与上下文长度
model_metadata.py
维护每个模型的上下文长度信息:
DEFAULT_CONTEXT_LENGTHS = {
"anthropic/claude-opus-4.6": 200000,
"openai/gpt-4o": 128000,
"google/gemini-2.5-pro": 1000000,
"meta-llama/llama-3-70b": 8192,
...
}Token 估算
def estimate_request_tokens_rough(messages, system_prompt, tools):
"""粗略估算请求 token 数"""
total = len(system_prompt) // 3.5 # 英文约 3.5 字符/token
for msg in messages:
total += len(str(msg.get("content", ""))) // 3.5
if tools:
total += len(json.dumps(tools)) // 3.5
return int(total)压缩触发
当估算 token 数接近模型上下文长度的阈值时,触发压缩:
threshold = context_length * 0.8 # 80% 作为阈值
if estimated_tokens >= threshold:
compress(messages)10.7 transports/ 目录详解
agent/transports/types.py 定义了三个核心数据类,是所有 Provider 适配器的统一输出格式:
@dataclass
class ToolCall:
id: Optional[str] # 协议标识符
name: str # 工具名称
arguments: str # JSON 字符串参数
provider_data: Optional[Dict] # 协议特定元数据
@dataclass
class Usage:
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
cached_tokens: int = 0 # 缓存命中 token(Anthropic)
@dataclass
class NormalizedResponse:
content: Optional[str]
tool_calls: Optional[List[ToolCall]]
finish_reason: str
reasoning: Optional[str]
usage: Optional[Usage]
provider_data: Optional[Dict] # 协议特定数据设计哲学:共享字段保持最小——只有所有下游消费者都需要的字段才放在顶层。协议特定的状态放在 provider_data 中。
例如:
- Codex:
{"call_id": "call_XXX", "response_item_id": "fc_XXX"} - Gemini:
{"extra_content": {"google": {"thought_signature": "..."}}} - Anthropic:
{"stop_reason": "end_turn", "cache_creation_input_tokens": 1234}
10.8 添加新 Provider
步骤 1:创建适配器文件
# agent/myllm_adapter.py
from agent.transports.types import NormalizedResponse, ToolCall, Usage
def call_myllm(model, messages, tools, **kwargs):
"""调用 MyLLM API 并归一化响应"""
raw = requests.post("https://myllm.api/v1/chat", json={
"model": model, "messages": messages, "tools": tools,
}).json()
tool_calls = None
if raw.get("tool_calls"):
tool_calls = [
ToolCall(id=tc["id"], name=tc["function"]["name"],
arguments=tc["function"]["arguments"])
for tc in raw["tool_calls"]
]
return NormalizedResponse(
content=raw.get("content"),
tool_calls=tool_calls,
finish_reason=raw.get("finish_reason", "stop"),
usage=Usage(
prompt_tokens=raw.get("usage", {}).get("prompt_tokens", 0),
completion_tokens=raw.get("usage", {}).get("completion_tokens", 0),
),
)步骤 2:在 run_agent.py 中注册路由
在 _make_api_call() 中添加 MyLLM 的检测和调用分支。
步骤 3:配置使用
hermes model # 选择 custom endpoint
# 或在 config.yaml 设置
# model: "myllm/my-model-v1"源码导航:
- 传输层 →
agent/transports/ - Anthropic →
agent/anthropic_adapter.py - Gemini →
agent/gemini_native_adapter.py - Bedrock →
agent/bedrock_adapter.py - 模型元数据 →
agent/model_metadata.py