BaseLLM 基类设计¶

BaseLLM 是所有 LLM 封装的抽象基类，定义了统一的核心接口和通用功能。它屏蔽了不同厂商 API 的差异，让上层应用可以用一致的方式调用各种大模型。

为什么需要 BaseLLM¶

在实际业务中，我们经常面临这样的困境：

成本控制：需要精确计算每次调用的 token 消耗，但不同厂商的计费规则差异巨大（如图片按"瓦片"计费 vs 按分辨率计费）
上下文管理：长对话容易超出模型限制，需要在合适的时机触发压缩，但各家的错误提示完全不同
厂商切换：业务初期用 GPT-4，后期想降级到 GPT-3.5 或切换到 Claude，但改造成本巨大
可观测性：需要追踪每次调用的耗时、成本、失败原因，但原始 API 返回的信息有限且格式不统一

BaseLLM 通过抽象层解决了这些问题，让上层代码专注于业务逻辑，而非厂商细节。

设计理念¶

三层抽象架构¶

BaseLLM 不是简单地把所有功能塞进一个类，而是通过分层设计逐步抽象：

最底层：具体厂商实现 - 直接调用厂商的 HTTP API - 处理厂商特有的数据格式转换 - 识别厂商特有的错误类型

中间层：ChatLLM / GenerationLLM - ChatLLM：处理对话式接口（支持 system/user/assistant 等角色） - GenerationLLM：处理补全式接口（只有单一文本输入输出） - 统一 Prompt 位置管理（system 消息放哪里、如何与 user 消息合并）

最上层：BaseLLM - Token 计算引擎（支持文本、图片、视频、音频、PDF） - 上下文压缩调度器（判断何时压缩、压缩到多小） - 媒体数据缓存（避免重复下载和计算） - OpenTelemetry 遥测埋点

这种设计让每层只关注自己的职责，新接入一个厂商时，只需实现底层和中间层的差异部分，上层功能全部继承。

职责分离的智慧¶

BaseLLM 遵循一个关键原则：LLM 层只负责调用，不负责上下文管理。

当上下文超出限制时，LLM 实现应该做的是： 1. 识别这是"上下文超长"错误（而不是网络错误或权限错误） 2. 从错误信息中提取当前大小和最大大小 3. 抛出标准化的 ContextTooLargeError 异常

然后由 Chain 层捕获异常，决定是压缩上下文、减少输入内容、还是直接报错。

这样设计的好处是： - LLM 实现专注于与厂商 API 打交道 - Chain 层拥有全局视角，可以制定更智能的压缩策略 - 错误处理逻辑集中在一处，而不是散落在每个 LLM 实现里

子类必须实现的核心方法¶

complete / async_complete¶

这是最重要的抽象方法，负责实际调用厂商 API 并返回结果。它接收的参数非常丰富： - current_input：当前用户输入 - conversation：历史对话列表 - elements：多模态内容（图片、视频等） - knowledge：知识库检索结果 - tools：可用的工具列表 - intermediate_msgs：中间步骤的消息（用于工具调用链） - response_format：期望的响应格式（纯文本 / JSON / 结构化数据）

实现时需要特别注意的是：当检测到上下文超长错误时，不要在 LLM 层做任何重试或压缩，而是抛出 ContextTooLargeError 异常，让 Chain 层处理。

construct_request_params¶

这个方法的任务是将 TFRobot 内部的数据结构"翻译"成厂商 API 能理解的格式。这通常涉及： - 将内部的消息对象转换为厂商的消息格式（如 {"role": "user", "content": "..."}） - 将工具列表转换为厂商的 function calling 格式 - 处理响应格式参数（JSON mode、JSON Schema 等）

每个厂商的格式都不一样，所以这个方法是接入新 LLM 时最需要定制的地方。

construct_llm_result¶

与 construct_request_params 相反，这个方法负责解析厂商的响应，提取出： - 模型生成的文本内容 - Token 使用情况（输入、输出、总计） - 工具调用信息（如果模型决定调用工具） - 原始响应（用于调试）

最终返回统一的 LLMResult 对象，让上层代码不需要关心是哪个厂商的模型。

reformat_request_msg_to_api（仅 ChatLLM）¶

这个方法处理单条消息的格式转换。它的特殊性在于： - 一个内部消息可能对应多个厂商消息（如包含多张图片的用户消息，某些厂商需要拆分成多条） - 使用泛型 ProviderMessageType，让每个子类定义自己的消息类型结构 - 处理特殊的角色映射（如 Anthropic 不支持 system 角色，需要将其合并到第一条 user 消息中）

extract_context_size_from_error¶

这是一个"救火"方法：当上下文超长时，从厂商的错误信息中提取两个关键数字： - 当前使用的 token 数量 - 模型的最大 token 限制

这两个数字会传递给 Chain 层的压缩逻辑，帮助它决定压缩到多小才合适。不同厂商的错误信息格式差异很大，有的直接返回 JSON，有的需要从文本中正则提取。

BaseLLM 提供的通用能力¶

Token 计算引擎¶

Token 计算是成本控制的基础，但不同厂商的计费规则差异巨大： - OpenAI：文本按 tiktoken 计算，图片按 512×512 的"瓦片"计费 - Claude：文本按自身规则计算，图片按分辨率计费 - Gemini：视频和音频有特殊的计费公式

BaseLLM 的 calculate_token_count 方法通过类型分发，自动识别输入类型并应用正确的计算规则：

纯文本：调用模型对应的 tokenizer（如 GPT-4 用 tiktoken）
消息对象：递归计算消息中所有内容的 token（role + content）
多模态元素：
图片：先下载图片，计算尺寸，按厂商规则切割成"瓦片"，再计费
视频：根据时长计算（Gemini 的公式是 258 + 258 × 秒数）
音频：根据时长计算（GPT-4o 的公式是 32 × 秒数）
PDF：按页数估算（默认 500 tokens/页）

计算结果会被缓存（见下文媒体缓存系统），避免重复计算。

智能上下文压缩¶

长对话的上下文压缩是保证连续对话体验的关键。BaseLLM 的 collapse_context 方法实现了两阶段压缩策略：

阶段一：快速机械压缩

当检测到上下文超出限制时，首先使用 Splitter 进行机械压缩。这个阶段的思路是"能切就切，不思考"： - 优先压缩历史对话（conversation） - 其次压缩多模态内容（elements） - 最后压缩知识库内容（knowledge） - 尽可能保留工具调用的中间消息（intermediate_msgs），因为它们通常包含最新和最重要的信息

阶段二：LLM 智能摘要

如果机械压缩后仍然超出限制，或者压缩后的内容仍然过多，就会调用 LLM 自身生成摘要。这个阶段的思路是"让模型自己决定保留什么"： - 构建一个多语言的压缩提示（中文或英文，取决于 locale 配置） - 调用 LLM，让它生成上下文的摘要 - 清空所有历史内容，只保留摘要作为新的 system 消息

递归保护

压缩过程可能会失败（如摘要本身太长），所以实现了递归保护： - 如果 LLM 调用仍然超出限制，目标大小衰减 10%（to_size × 0.9） - 最多递归 5 次 - 如果仍然失败，抛出异常让上层处理

媒体缓存系统¶

多模态内容的处理非常耗时（下载图片、计算尺寸、切割瓦片），而且这些操作在多次调用中是重复的。BaseLLM 使用 TTLCache 缓存媒体数据，包含：

媒体类型：image / video / audio / pdf
字节数据：下载后的原始数据和 MIME 类型
尺寸信息：宽高、瓦片数量
Token 数量：计算好的 token 消耗
缓存时间：用于 TTL 过期

缓存的 key 格式是 "{media_type}:{uri}:{param_hash}"，其中 param_hash 包含了可能影响计算的参数（如图片的 detail 参数）。

每个 LLM 实例都有独立的缓存，最大 5000 条目，24 小时过期。这样即使多个线程使用同一个 LLM 实例，也不会互相干扰。

遥测追踪¶

BaseLLM 通过 OpenTelemetry 自动记录每次 LLM 调用的完整生命周期： - BEFORE_LLM_GENERATE：调用开始，记录模型名称、输入 token 数 - AFTER_LLM_GENERATE：调用成功，记录输出 token 数、耗时、成本 - LLM_GENERATE_RAISE：抛出异常，记录异常类型和错误信息 - LLM_GENERATE_ABORT：调用中止，记录中止原因

这些追踪数据会被发送到配置的 OTel 后端（如 Jaeger），用于性能分析、成本核算、问题排查。

工具调用消息优化¶

在使用 Function Calling 时，中间消息会快速累积（调用工具 → 工具返回 → 继续调用下一个工具），这些消息会占用大量 token。

BaseLLM 的 optimize_map_reduce_messages 方法识别并优化这种模式：

Map-Reduce 模式识别

当工具调用形成"调用-返回-再调用-再返回"的链条时，这个方法会： 1. 识别所有 Map 消息（模型调用工具的消息） 2. 识别所有 Reduce 消息（工具返回结果的消息） 3. 将多个 Map 消息的返回结果融合到一个 Reduce 消息中

工具调用树支持

对于嵌套的工具调用（如 A 调用 B，B 调用 C），方法会构建调用树，只保留必要的路径节点，删除中间的重复信息。

这个优化可以节省 30-50% 的 token 消耗，特别是在复杂的 Agent 场景中。

关键配置参数¶

基础配置¶

参数	作用	典型值
`name`	模型名称，用于日志和遥测	`gpt-4o`, `claude-3-5-sonnet-20241022`
`input_price` / `output_price`	Token 单价（元/千 tokens），用于成本计算	GPT-4o: 0.025 / 0.125
`context_window`	上下文窗口大小，用于预测是否需要压缩	GPT-4o: 128000
`locale`	内部提示的语言（中文/英文），影响压缩提示等	`Locale.ZH` 或 `Locale.EN`

response_format：控制模型输出格式¶

这个参数决定了模型应该如何返回结果：

text（默认）：自由文本，模型可以返回任何内容
json_object：模型必须返回合法的 JSON，但不能指定 Schema
json_schema：模型必须返回符合指定 Schema 的 JSON（如果模型支持）

使用场景举例：

场景一：需要结构化数据抽取

response_format = {
    "type": "json_schema",
    "json_schema": {
        "name": "user_info",
        "schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
                "email": {"type": "string"}
            },
            "required": ["name", "age"]
        }
    }
}

这样配置后，模型会保证返回包含 name、age、email 字段的 JSON，可以直接解析使用。

场景二：需要 JSON 但 Schema 很复杂

response_format = {
    "type": "json_object",
    "examples": [
        '{"name": "张三", "age": 25, "hobbies": ["读书", "游泳"]}',
        '{"name": "李四", "age": 30, "hobbies": ["登山", "摄影"]}'
    ]
}

通过 examples 提供 few-shot 示例，引导模型输出符合期望的 JSON 格式。

tool_filter：工具过滤表达式（重点）¶

为什么需要 tool_filter？

在实际业务中，我们经常遇到这样的问题： - 模型被提供了 100 个工具，但当前场景只需要其中 3 个 - 某些工具只对特定用户开放（如管理员工具） - 不同阶段需要不同的工具集（如"搜索"阶段只需要搜索工具，"操作"阶段需要执行工具） - 工具太多会导致模型"分心"，选择错误的工具或产生幻觉

tool_filter 参数通过表达式语言，让上层代码可以在运行时动态过滤工具列表。

语法规则

tool_filter 使用类似 JSONPath 的表达式语法： - tag:search：只保留带有 search 标签的工具 - category:math：只保留分类为 math 的工具 - !tag:admin：排除带有 admin 标签的工具 - tag:search | tag:analysis：保留带 search 或 analysis 标签的工具 - tag:search & category:web：保留同时满足条件的工具 - tag:search & !tag:beta：组合过滤，搜索工具但不包括测试版

实际应用场景

场景一：分阶段的 Agent

假设我们实现了一个电商助手，包含三个阶段：

# 搜索阶段：只需要搜索类工具
llm = ChatLLM(
    name="gpt-4o",
    tool_filter="tag:search | tag:compare"
)
# 可用工具：search_products, compare_prices, search_reviews

# 推荐阶段：需要分析和推荐工具
llm = ChatLLM(
    name="gpt-4o",
    tool_filter="tag:recommend | tag:analysis"
)
# 可用工具：analyze_preferences, recommend_products, check_inventory

# 下单阶段：需要执行类工具
llm = ChatLLM(
    name="gpt-4o",
    tool_filter="tag:order | tag:payment"
)
# 可用工具：create_order, process_payment, send_confirmation

通过分阶段过滤，模型在每个阶段都能专注于当前任务相关的工具，减少混淆和错误调用。

场景二：用户权限控制

# 普通用户：排除管理员工具
regular_user_llm = ChatLLM(
    name="gpt-4o",
    tool_filter="!tag:admin & !tag:internal"
)

# 管理员：可以使用所有工具
admin_llm = ChatLLM(
    name="gpt-4o",
    tool_filter=None  # 不过滤
)

通过给工具打上 tag:admin 标签，可以轻松实现权限控制，无需在代码中维护工具白名单。

场景三：A/B 测试

# 控制组：使用旧版搜索工具
control_llm = ChatLLM(
    name="gpt-4o",
    tool_filter="tag:search & !tag:v2"
)

# 实验组：使用新版搜索工具
experiment_llm = ChatLLM(
    name="gpt-4o",
    tool_filter="tag:search & tag:v2"
)

通过给不同版本的打上不同标签，可以轻松进行功能灰度和 A/B 测试。

最佳实践

工具标签设计：给每个工具打上清晰的标签（category、功能、权限、版本等）
渐进式过滤：从宽泛到具体，先用 category 过滤，再用 tag 精确过滤
性能考虑：过滤是在调用前完成的，不会影响模型推理速度
组合使用：可以用 & 和 | 组合多个条件，构建复杂的过滤逻辑

如何扩展新 LLM¶

接入一个新的 LLM 提供商，本质上就是实现 BaseLLM 定义的一系列抽象方法。整体流程如下：

第一步：选择基类¶

如果模型是对话式的（有 system/user/assistant 角色），继承 ChatLLM。如果模型是补全式的（只有单一文本输入输出），继承 GenerationLLM。

大多数现代模型（GPT-4、Claude、Gemini）都是对话式的，所以通常继承 ChatLLM。

第二步：定义模型元数据¶

在类级别定义： - name：模型名称（如 gpt-4o），用于日志和配置识别 - input_price / output_price：Token 单价，用于成本计算 - api_key：厂商 API 密钥

这些信息可以通过 whosellm 工具自动获取，也可以手动配置。

第三步：初始化客户端¶

重写 model_post_init 方法，在这个方法中： - 创建厂商的 HTTP 客户端（如 OpenAI 的 openai.OpenAI()） - 设置认证信息（API Key、Endpoint 等） - 配置客户端参数（超时、重试等）

这个方法会在 Pydantic 模型初始化后自动调用，确保所有配置都已就绪。

第四步：实现核心方法¶

construct_request_params

这是最复杂的方法，需要完成以下任务： 1. 调用父类的 format_to_request_msgs，获取格式化后的消息列表 2. 遍历每条消息，调用 reformat_request_msg_to_api 转换为厂商格式 3. 处理工具列表（如果有），转换为厂商的 function calling 格式 4. 添加厂商特有的参数（如温度、top_p 等） 5. 返回完整的请求参数字典

complete

这个方法负责实际调用厂商 API： 1. 调用 construct_request_params 构造请求参数 2. 调用厂商的 HTTP 方法（如 client.chat.completions.create()） 3. 使用 try-except 捕获异常，特别是识别上下文超长错误 4. 调用 construct_llm_result 解析响应 5. 返回 LLMResult 对象

reformat_request_msg_to_api

这个方法处理单条消息的转换： - 将 TFRobot 的 LLMSystemMessage、LLMUserMessage、LLMAssistantMessage 转换为厂商的消息格式 - 处理多模态内容（如图片 URL、视频数据） - 处理特殊的角色映射（如将 system 消息合并到第一条 user 消息）

construct_llm_result

这个方法负责解析厂商的响应： - 提取生成的文本内容 - 提取 token 使用情况（如果厂商返回） - 提取工具调用信息（如果模型调用了工具） - 构造 LLMResult 对象并返回

extract_context_size_from_error

这个方法处理上下文超长错误： - 从厂商的异常信息中提取当前 token 数量 - 提取模型的最大 token 限制 - 返回这两个数字的元组

不同厂商的错误格式差异很大，有的返回 JSON，有的需要正则解析，这是最需要定制化的部分。

第五步：处理多模态内容（可选）¶

如果模型支持多模态（如 GPT-4o、Gemini Pro Vision），可以重写 _estimate_*_tokens 方法：

_estimate_image_tokens：计算图片的 token 消耗（考虑瓦片、分辨率等）
_estimate_video_tokens：计算视频的 token 消耗（考虑时长、帧率等）
_estimate_audio_tokens：计算音频的 token 消耗（考虑时长、采样率等）

这些方法会被 calculate_token_count 自动调用，用于成本预测和压缩决策。

第六步：添加重试和容错（推荐）¶

使用 tenacity 库为 complete 方法添加重试逻辑： - 对网络错误、超时错误进行重试 - 使用指数退避策略（等待时间逐渐增加） - 限制最大重试次数（如 3 次） - 对上下文超长错误不重试（直接抛出异常）

第七步：支持异步（推荐）¶

同时实现同步和异步版本的 complete 方法： - 同步方法使用同步的 HTTP 客户端 - 异步方法使用异步的 HTTP 客户端（如 httpx.AsyncClient） - 两个方法共享相同的逻辑，只是 I/O 操作不同

这样可以在异步环境中使用（如 FastAPI），不会阻塞事件循环。

实现建议¶

从测试入手：先写一个简单的测试，调用模型的 complete 方法，看看能否正常返回结果
逐步完善：先实现基本功能，再添加多模态、重试、异步等高级特性
参考现有实现：GPT-4、Claude 的实现已经很完善，可以作为参考模板
使用厂商 SDK：不要自己写 HTTP 请求，使用厂商提供的 Python SDK（如 openai、anthropic）
正确处理异常：特别关注上下文超长错误，这是最容易出问题的地方
添加日志和追踪：使用 OpenTelemetry 记录每次调用，便于问题排查

最佳实践总结¶

使用层面¶

善用 tool_filter：不要一股脑把所有工具都丢给模型，根据场景动态过滤工具集，这样能显著提高准确率
合理设置 response_format：需要结构化输出时优先使用 json_schema，其次是 json_object + examples
关注 token 成本：调用前使用 calculate_token_count 预估成本，避免意外超支
配置 locale：如果业务主要是中文，设置 locale=Locale.ZH，让内部提示（如压缩提示）使用中文

实现层面¶

复用父类能力：token 计算、上下文压缩、媒体缓存等功能已经在 BaseLLM 中实现，不要重复造轮子
正确处理异常：特别是上下文超长错误，一定要抛出 ContextTooLargeError 而不是自己在 LLM 层重试
使用厂商 SDK：不要自己写 HTTP 请求代码，厂商提供的 SDK 已经处理了认证、重试、流式响应等细节
异步和同步同步实现：两个方法应该共享相同的逻辑，只是 I/O 层面不同
添加测试：至少要有一个端到端测试，验证 complete 方法能正常工作

可观测性¶

利用 OpenTelemetry：BaseLLM 已经自动埋点，只需要配置 OTel 后端（如 Jaeger）
关注关键指标：调用耗时、token 消耗、成本、错误率
追踪上下文压缩：当压缩发生时，会记录压缩前后的 token 数量，这对于优化很有价值