跳到主要内容

多供应商 LLM 抽象:统一 chat 接口、消息/工具翻译与重试

30 秒导读: Trae Agent 的 agent 循环(第 1 章)只知道一件事——"给我一份消息列表和一组工具,还我一个带 contenttool_calls 的回复"。至于对面是 Anthropic、OpenAI 还是 Ollama,循环完全不在乎。这层"不在乎"就是本章讲的东西:trae_agent/utils/llm_clients/ 里一套中立数据模型 + 每家一个客户端子类,把 7 家供应商的消息格式差异、工具 schema 差异、tool-call 拆解差异,全部关进各自 chat() 的内部。

本章不讲 agent 怎么循环(那是第 1 章),也不讲工具本身怎么执行(那是第 2 章)。本章只讲一件事:agent 如何用一套代码对接 7 家供应商


1. 这是什么(零基础也能懂)

一句话定义

这是一个**"翻译中间层":上层 agent 用一种统一的中立格式**说话,这层负责把它翻译成每家供应商各自的方言,发出去,再把各家五花八门的回复翻译回同一种中立格式交还给上层。

它解决什么问题

不同 LLM 供应商的 API 长得完全不一样,尤其在两件事上:

  • 消息格式——同样是"工具执行结果",Anthropic 要你塞进一个 role="user" 消息里的 tool_result content block;OpenAI 系要一条独立的 role="tool" 消息。
  • 工具描述(tool schema)——同样是"我有一个 bash 工具",Anthropic 对某些内置工具有原生类型(bash_20250124),OpenAI 要包成 type="function" 的 JSON Schema,而且 strict 模式下所有参数都得进 required

如果不做抽象,agent 循环里就会撒满 if provider == "anthropic": ... elif provider == "openai": ...,加一家新供应商就要改一圈。Trae 的做法是:把这些 if 全部收进 llm_clients/,循环只面对一个干净的 LLMClient.chat()

它能做什么

  • 一套代码对接 7 家:OpenAI、Anthropic、Azure、Ollama、OpenRouter、Doubao(豆包)、Google。
  • 懒加载:构造时只 import 你实际选的那一家的客户端,其余不碰。
  • 统一的 usage 统计、统一的重试退避、统一的轨迹记录(第 6 章)。

用起来什么样

上层拿到 LLMClient 后,一次调用就是这样(agent 循环里的真实用法):

# 示意,非源码:agent 侧只见这一个统一接口
client = LLMClient(model_config) # 按 provider 懒加载具体子类
client.set_trajectory_recorder(recorder) # 挂上轨迹记录
response = client.chat( # 中立入参:LLMMessage 列表 + Tool 列表
messages=[LLMMessage(role="user", content="修一下这个 bug")],
model_config=model_config,
tools=[bash_tool, edit_tool],
reuse_history=True, # 复用累积的对话历史
)
# 中立出参:content 是文本,tool_calls 是要执行的工具调用
for call in response.tool_calls or []:
... # 交给第 2 章的 ToolExecutor 去跑

注意:整段代码里没有一处出现供应商名字。 这就是抽象的目标。

一句话直觉

把它当成一个"多国语言同声传译"。 会议桌上层(agent)只讲一种"公司内部语"(LLMMessage),传译员(具体 client)负责把它翻成对面客人的母语(Anthropic / OpenAI 格式),再把对面的回答翻回内部语(LLMResponse)。换个客人只是换个传译员,会议流程一个字不用改。


2. 顶层全景(它大概怎么转)

一张图:一次 chat 从中立格式到某家 API 再回来

怎么读:从上到下是一次 chat() 的数据流。左边是中立世界(agent 只认这些类型),中间竖线是翻译边界,右边是某家供应商的方言

中立世界(agent 只见这层) │ 翻译边界(具体 client 内部) │ 供应商方言
│ │
list[LLMMessage] ──────────────────────┼──► parse_messages() │
(role/content/tool_call/tool_result) │ 按 role 和字段分派翻译 ───────┼──► MessageParam[] (Anthropic)
│ │ 或 ChatCompletionMessageParam[]
list[Tool] ─────────────────────────────┼──► tool_schemas 生成 │
(第 2 章的工具对象) │ 内置工具→原生类型 / 其余→通用 ─┼──► ToolUnionParam[] / ToolParam[]
│ │
│ retry_with(...) 包一层重试 ───┼──► client.messages.create(...)
│ │ │
│ 拆 response.content ◄──────────┼─────────┘
LLMResponse ◄──────────────────────────┼── text→content / tool_use→ToolCall│
(content / tool_calls / usage) │ 记 usage + 写 trajectory │

部件一句话职责

部件干什么在哪个文件
LLMProvider 枚举列出 7 家供应商的字符串标识llm_client.py:15
LLMClient门面:按 provider match 懒加载具体子类,再把调用转发过去llm_client.py:27
BaseLLMClient抽象基类:定死 chat() / set_chat_history() 契约base_client.py:13
LLMMessage / LLMResponse / LLMUsage三个中立数据模型,抽象的"公司内部语"llm_basics.py:11,21,45
AnthropicClientAnthropic 方言的完整翻译实现(本章主线)anthropic_client.py:19
OpenAICompatibleClientAzure / OpenRouter / Doubao 共用的 OpenAI 兼容翻译层openai_compatible_base.py:65
retry_with给任何 API 调用套上"失败随机退避重试"retry_utils.py:13
Tool.get_input_schema()model_provider 分支,生成各家能吃的参数 schematools/base.py:127

主线走一遍(高层,不进代码)

  1. agent 拿 model_config 造一个 LLMClient → 里面按 provider 懒加载出真正干活的子类(如 AnthropicClient)。
  2. agent 调 client.chat(messages, ..., tools)
  3. 子类内部:先 parse_messages() 把中立消息翻成方言消息;再把 tools 翻成方言的 tool schema。
  4. retry_with 包一层,真正调供应商 API。
  5. 拿回原始响应,拆开 content block:文本拼进 content,工具调用拆成 ToolCall
  6. 统计 usage、写轨迹,返回一个中立的 LLMResponse

3. 核心机制一:门面 + 懒加载(LLMClient + LLMProvider)

它要解决的小问题

7 家供应商各有一个客户端类,每个都 import 一堆重量级 SDK(anthropicopenaigoogle-genai…)。如果构造 LLMClient 时把 7 个全 import 进来,既慢又可能因为没装某个 SDK 而报错——而用户其实只用一家。

思路

把 import 语句写进 match 的每个分支里。 Python 的 import 是执行到才生效的语句,所以只有命中的那个分支的 SDK 才被真正加载。

真实实现

LLMProvider 就是一个字符串枚举,把配置里的 provider 字段收敛成有限集合:

# llm_client.py:15
class LLMProvider(Enum):
OPENAI = "openai"
ANTHROPIC = "anthropic"
AZURE = "azure"
OLLAMA = "ollama"
OPENROUTER = "openrouter"
DOUBAO = "doubao"
GOOGLE = "google"

构造时按枚举 match,import 写在分支内,命中谁才加载谁:

# llm_client.py:34
match self.provider:
case LLMProvider.OPENAI:
from .openai_client import OpenAIClient
self.client: BaseLLMClient = OpenAIClient(model_config)
case LLMProvider.ANTHROPIC:
from .anthropic_client import AnthropicClient
self.client = AnthropicClient(model_config)
# ... 其余 5 家同构

注意 self.client 的类型标注是 BaseLLMClient——门面只依赖抽象基类,不依赖任何具体子类。这正是抽象的落点:LLMClient.chat()(llm_client.py:72)只是一句 return self.client.chat(...),把调用透明转发给多态的子类。

关键细节

LLMClient 本身几乎没有逻辑,它是一个薄门面(facade):chatset_chat_historyset_trajectory_recordersupports_tool_calling 全是转发。真正的多态发生在 self.client 上。supports_tool_calling(llm_client.py:82)还多做了一层 hasattr 防御,子类没实现也不会炸。


4. 核心机制二:中立数据模型(llm_basics.py)

它要解决的小问题

要让 7 家共用一个接口,先得有一种谁都不属于、谁都能翻译到/翻译回的中间语言。这就是 llm_basics.py 里那三个 @dataclass

三个模型各管一段

模型管什么关键字段
LLMMessage一条输入消息(可以是文本、工具调用、或工具结果)role / content / tool_call / tool_result
LLMResponse一次输出回复content / tool_calls / usage / finish_reason
LLMUsagetoken 用量统计input_tokens / output_tokens / cache_* / reasoning_tokens

LLMMessage 的巧妙在于四个字段用组合表达"消息种类",而不是搞一个类型枚举:

# llm_basics.py:11
@dataclass
class LLMMessage:
role: str
content: str | None = None
tool_call: ToolCall | None = None # 有值 = 这是一条"我要调工具"的消息
tool_result: ToolResult | None = None # 有值 = 这是一条"工具跑完了"的消息

翻译时就靠"哪个字段有值"来分派(见 §5 的 parse_messages)。ToolCall / ToolResult 本身来自第 2 章tools/base.py,是跨 LLM 层和工具层共享的类型。

LLMUsage 的可加性

LLMUsage 重载了 __add__(llm_basics.py:30),于是多轮对话的用量可以直接 total = total + response.usage 逐字段累加。它还专门留了 cache_creation_input_tokens / cache_read_input_tokens(Anthropic 的 prompt 缓存)和 reasoning_tokens(推理模型),让不同供应商的特殊计费口径都能落进同一个结构。


5. 核心机制三:以 Anthropic 为主线走通一次 chat

这是本章的重头戏。我们跟着 AnthropicClient.chat()(anthropic_client.py:53)从头走一遍,看一次调用里发生的每一步翻译。

5.1 入口契约:BaseLLMClient

每个客户端都继承 BaseLLMClient(base_client.py:13),它用 @abstractmethod 定死两件必须实现的事——chat()set_chat_history()——并在 __init__ 里从 model_config.model_provider 抽出 api_key / base_url / api_version 三个通用字段。supports_tool_calling() 有个默认实现(读 model_config.supports_tool_calling),子类可覆盖。

5.2 第一步:parse_messages —— 中立消息翻成 Anthropic 消息

parse_messages(anthropic_client.py:155)是翻译的核心。它遍历每条 LLMMessage,按"哪个字段有值"分派成不同的 Anthropic MessageParam:

中立消息(判据)翻成 Anthropic 的什么关键代码
role == "system"不进消息列表,单独存到 self.system_message(create 时作 system= 参数)anthropic_client.py:159
tool_result 有值role="user" 消息,content 是一个 tool_result blockanthropic_client.py:161
tool_call 有值role="assistant" 消息,content 是一个 tool_use blockanthropic_client.py:168
普通 user/assistantrole=原样,content 是纯字符串anthropic_client.py:174

这里藏着 Anthropic 方言的第一个"坑":system 提示不是一条消息,而是 API 的独立参数。所以翻译时要把它从消息流里"抽走",存进 self.system_message:

# anthropic_client.py:159
if msg.role == "system":
self.system_message = msg.content if msg.content else anthropic.NOT_GIVEN

工具结果的翻译还顺手做了容错(parse_tool_call_result,anthropic_client.py:199):把 resulterror 拼成一段文本,is_error 置为 not success;如果工具失败又没给任何细节,补一句默认 "Tool execution failed without providing error details.",避免给模型一个空 block。

5.3 第二步:tool_schemas 生成 —— 内置工具走原生类型

工具翻译(anthropic_client.py:73)是 Anthropic 相比 OpenAI 系最不一样的地方:Anthropic 对某些工具提供服务端原生实现,你只要报一个类型名,模型就知道怎么用,不需要你描述 schema。

# anthropic_client.py:76,83,88 —— 按工具名分三档翻译
if tool.name == "str_replace_based_edit_tool":
tool_schemas.append(TextEditor20250429( # 原生文本编辑器类型
name="str_replace_based_edit_tool", type="text_editor_20250429"))
elif tool.name == "bash":
tool_schemas.append(anthropic.types.ToolBash20250124Param( # 原生 bash 类型
name="bash", type="bash_20250124"))
else:
tool_schemas.append(anthropic.types.ToolParam( # 通用:才需要描述 + schema
name=tool.name, description=tool.description,
input_schema=tool.get_input_schema()))

三档翻译:

工具名翻成为什么
str_replace_based_edit_toolTextEditor20250429(原生)Anthropic 有服务端文本编辑器,不用报 schema
bashToolBash20250124Param(原生)同上,原生 bash 工具
其余所有ToolParam + get_input_schema()普通函数工具,得完整描述参数

对比一下 OpenAICompatibleClient(openai_compatible_base.py:126):它对所有工具一视同仁,统统包成 ChatCompletionToolParam(type="function") + get_input_schema(),没有原生工具这一说。这个差异正是抽象层要吸收的东西之一。

5.4 第三步:发请求(套重试)

真正的 API 调用抽成了一个私有方法 _create_anthropic_response(anthropic_client.py:36),它把 self.message_historyself.system_messagetool_schemas 和采样参数(temperature/top_p/top_k)一起交给 client.messages.create。这个方法不直接调,而是先用 retry_with 包一层(见 §6)再调:

# anthropic_client.py:97
retry_decorator = retry_with(
func=self._create_anthropic_response,
provider_name="Anthropic",
max_retries=model_config.max_retries,
)
response = retry_decorator(model_config, tool_schemas)

5.5 第四步:拆响应 —— content block 拆成 content + ToolCall

Anthropic 的回复是一个 content block 列表,可能混着文本块和工具调用块。拆解逻辑(anthropic_client.py:108)遍历它们,分别处理:

# anthropic_client.py:108
for content_block in response.content:
if content_block.type == "text":
content += content_block.text # 文本→拼进 content
self.message_history.append(...) # 同时回写历史(assistant 文本)
elif content_block.type == "tool_use":
tool_calls.append(ToolCall( # 工具块→拆成中立 ToolCall
call_id=content_block.id,
name=content_block.name,
arguments=content_block.input))
self.message_history.append(...) # 回写历史(assistant tool_use)

拆完组装成中立的 LLMResponse(anthropic_client.py:135):content 是拼起来的文本,tool_calls 是那一串 ToolCall(空则给 None),finish_reasonresponse.stop_reason。至此,上层拿到的就是与供应商无关的统一结构。

5.6 message_history 复用(reuse_history)

chat() 有个 reuse_history 开关(默认 True)。它决定这次的新消息是追加到累积历史后面,还是把历史整个替换掉:

# anthropic_client.py:65
self.message_history = (
self.message_history + anthropic_messages if reuse_history else anthropic_messages
)
  • reuse_history=True(多轮对话常态):新消息接在已有历史后,模型看得到完整上下文。
  • reuse_history=False:丢掉旧历史,只用这次的消息(比如做一次性的独立调用)。

而且 §5.5 里拆响应时,assistant 的文本块和 tool_use 块都会被回写进 self.message_history——这样下一轮 chat 就自带了"上一轮模型说了啥、调了啥工具"。set_chat_history()(anthropic_client.py:31)则是另一条路:直接用一批 LLMMessage 预置历史。

5.7 usage 统计与 trajectory 记录

response.usage 抽出四个 token 计数(含 Anthropic 特有的两个 cache 字段)填进 LLMUsage(anthropic_client.py:126)。最后,若挂了轨迹记录器,就把这次交互原样落盘:

# anthropic_client.py:144
if self.trajectory_recorder:
self.trajectory_recorder.record_llm_interaction(
messages=messages, response=llm_response,
provider="anthropic", model=model_config.model, tools=tools)

record_llm_interaction 的签名见 trajectory_recorder.py:77;轨迹记录的整体机制属于第 6 章。这里的要点是:每个客户端在 chat 结尾都做同一件事——把中立的 messages + LLMResponse + provider 名交给记录器,于是不管哪家供应商,轨迹格式都是统一的。


6. 核心机制四:retry_with 的重试包装

它要解决的小问题

LLM API 会因限流、超时、网络抖动而偶发失败。每家客户端都需要"失败了等一会儿再试",不应各写一遍。

思路

写一个高阶函数:传入要调的函数,返回一个"带重试的同名函数"。谁都能用一行套上。

真实实现

# retry_utils.py:13
def retry_with(func, provider_name="OpenAI", max_retries=3):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries:
raise # 最后一次仍失败:抛出
sleep_time = random.randint(3, 30) # 随机退避 3-30 秒
print(f"{provider_name} API call failed: {e}. ...")
time.sleep(sleep_time)

几个设计点:

  • 随机退避(jitter):每次睡 random.randint(3, 30) 秒,而非固定间隔——避免多个并发请求同时重试造成"惊群"。(注意:这里是纯随机,不是指数退避。)
  • 最后一次不睡直接抛:attempt == max_retriesraise,把真实异常暴露给上层。
  • catch 所有 Exception:不区分错误类型,任何异常都重试(简单粗暴,但对"临时故障"够用)。

retry_with 在 Anthropic 客户端(anthropic_client.py:97)和 OpenAI 兼容层(openai_compatible_base.py:142)里都被用来包裹那个私有的 _create_*_response 方法。provider_name 只用于日志文案。


7. openai_compatible_base.py:三家共用的 OpenAI 兼容层

它为什么存在

有些供应商的 API 与 OpenAI 的 chat.completions 协议兼容,只在"用哪个 base_url、要不要额外 header、哪些模型支持工具"上有差别。为它们各写一份完整 chat() 是重复劳动。

它怎么复用

OpenAICompatibleClient(openai_compatible_base.py:65)把完整的 chat 流程(parse_messages、tool schema、重试、拆响应、写轨迹)写死一遍;把会变的部分抽成一个抽象类 ProviderConfig(openai_compatible_base.py:34),只留四个钩子:

ProviderConfig 钩子各家填什么
create_client()用什么 base_url / 客户端类型建 openai 客户端
get_service_name()重试日志里显示的名字
get_provider_name()写轨迹用的 provider 标识
get_extra_headers()额外 HTTP header(如 OpenRouter 的 HTTP-Referer)

于是每家只需一个十几行的小文件。以 Doubao 为例(doubao_client.py),DoubaoProvider 填好四个钩子,DoubaoClient 只是 super().__init__(model_config, DoubaoProvider());OpenRouter(openrouter_client.py)类似,额外在构造时兜底一个默认 base_url、并从环境变量读站点 header;Azure(azure_client.py)则在 create_client 里改用 openai.AzureOpenAI

一个要澄清的事实

本层实际被三家复用:Azure、OpenRouter、Doubao(都 class XxxClient(OpenAICompatibleClient))。而 OpenAI 自己和 Ollama 并不走这层——它俩直接继承 BaseLLMClient:OpenAIClient 用的是较新的 Responses API(openai_client.py:48responses.create),OllamaClient 有自己的 chat.completions 实现。这是读代码时容易想当然搞错的一点(inferred:从各文件的类声明和调用点核对得出)。

OpenAI 系的消息翻译差异

同样在 parse_messages(openai_compatible_base.py:217)里,可以直观看到与 Anthropic 的分野:

中立消息Anthropic 译法OpenAI 兼容译法
system抽成 system= 参数一条 role="system" 消息
tool_resultrole="user" 里塞 tool_result block独立 role="tool" 消息(tool_call_id 关联)
tool_callrole="assistant" 里塞 tool_use blockrole="function" 消息

这三行差异,正是"抽象层存在的理由"——上层的 LLMMessage 一个字没变,变的全在各自 parse_messages 内部。


8. 各家 schema 差异如何被 get_input_schema 吸收

工具的参数 schema 还有一层供应商差异,不在 llm_clients 里,而在第 2 章Tool.get_input_schema()(tools/base.py:127)里按 model_provider 分支消化。这里点出与本章直接相关的部分。

同一个工具,给不同供应商生成的 schema 不同,差异集中在 OpenAI 的 strict 模式要求:

# tools/base.py:144 —— OpenAI 分支:所有参数都进 required,可选的改成 nullable
if self.model_provider == "openai":
required.append(param.name)
if not param.required:
current_type = param_schema["type"]
if isinstance(current_type, str):
param_schema["type"] = [current_type, "null"] # 可选 → 类型加 "null"

OpenAI 分支还会在对象类型和顶层 schema 上补 additionalProperties: false(tools/base.py:162,172)。其余供应商则只把 param.required 为真的参数放进 required(tools/base.py:152)。

要点: 工具本身(第 2 章)用 ToolParameter 这种中立方式声明参数;到底渲染成哪家的 JSON Schema,由 get_input_schemamodel_provider 决定。这样 llm_clients 层拿到的 input_schema 已经是"对味"的,AnthropicClient / OpenAICompatibleClient 直接塞进各自的 ToolParam 即可,无需再关心 strict 模式这类细节。


9. 边界与局限(诚实)

  • 重试是纯随机退避,不是指数退避(retry_utils.py:44),也不区分错误类型——4xx 参数错误和 429 限流会被一样地重试。对偶发故障够用,对结构性错误只是白等几轮再抛。
  • 重试内部用 print 打日志(retry_utils.py:46),不走结构化日志。
  • supports_tool_calling 的判断偏启发式:OpenRouter 靠模型名匹配一串已知 pattern(openrouter_client.py:48),Doubao 直接返回 True(doubao_client.py:36)——新模型可能误判。
  • 门面注释已过时:llm_client.py:4 的模块 docstring 仍写 "OpenAI, Anthropic, Azure, and OpenRouter",实际已支持 7 家。
  • OpenAI / Ollama 未复用兼容层:如 §7 所述,这两家各自实现,意味着"OpenAI 兼容"的通用改动不会自动惠及它们。

10. 横向对比(与兄弟章节)

  • 本章的产物 LLMResponse.tool_calls 正是第 1 章 agent 循环的燃料:循环拿到 tool_calls → 交给第 2 章ToolExecutor 执行 → 把 ToolResult 包成 LLMMessage(tool_result=...) → 再进本章 chat() 的下一轮。
  • 每次 chat() 结尾的 record_llm_interaction 属于第 6 章的轨迹与可观测性体系。
  • 工具参数 schema 的生成(§8)真正的家在第 2 章tools/base.py

11. 代码地图(导航索引)

主题文件路径符号名
供应商枚举trae_agent/utils/llm_clients/llm_client.pyLLMProvider
门面 + 懒加载trae_agent/utils/llm_clients/llm_client.pyLLMClient
抽象基类契约trae_agent/utils/llm_clients/base_client.pyBaseLLMClient
中立消息/回复/用量trae_agent/utils/llm_clients/llm_basics.pyLLMMessage / LLMResponse / LLMUsage
Anthropic 一次 chattrae_agent/utils/llm_clients/anthropic_client.pyAnthropicClient.chat
消息翻译(Anthropic)trae_agent/utils/llm_clients/anthropic_client.pyAnthropicClient.parse_messages
工具结果翻译trae_agent/utils/llm_clients/anthropic_client.pyAnthropicClient.parse_tool_call_result
OpenAI 兼容共用层trae_agent/utils/llm_clients/openai_compatible_base.pyOpenAICompatibleClient / ProviderConfig
兼容层消息翻译trae_agent/utils/llm_clients/openai_compatible_base.pyOpenAICompatibleClient.parse_messages
供应商小配置(示例)trae_agent/utils/llm_clients/doubao_client.pyDoubaoProvider / DoubaoClient
重试退避trae_agent/utils/llm_clients/retry_utils.pyretry_with
工具 schema 按供应商分支trae_agent/tools/base.pyTool.get_input_schema